This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new de9d221da902 [SPARK-60416][SQL][SS] Allow Structured Streaming
Coverage in SQL Tests
de9d221da902 is described below
commit de9d221da902d7391fdb05ab349dd59ebdd46348
Author: Siying Dong <[email protected]>
AuthorDate: Wed Oct 28 17:44:15 2026 +0000
[SPARK-60416][SQL][SS] Allow Structured Streaming Coverage in SQL Tests
### What changes were proposed in this pull request?
This PR adds `StreamingSQLQueryTestSuite`, a `SQLQueryTestSuite` subclass
that lets SQL golden
tests exercise Structured Streaming flows.
The test harness now:
- parses `CREATE FLOW` statements and retains their streaming logical plans;
- recognizes test-only, case-insensitive `REFRESH FLOW` statements;
- runs the saved flow with `Trigger.AvailableNow` into an existing table
and reuses its checkpoint
across refreshes; and
- normalizes event-time watermark UUIDs in golden output.
It adds golden coverage for writing to a table, stream-static joins,
streaming unions, windowed
aggregations with watermarks, and multiple chained flows.
### Why are the changes needed?
SQL golden tests currently cannot execute a query containing streaming
sources because ordinary
batch execution rejects its streaming logical plan. Running the saved
`CREATE FLOW` query through
`writeStream` lets SQL tests validate Structured Streaming behavior and
inspect the resulting table
with regular SQL.
### Does this PR introduce _any_ user-facing change?
No. This only changes the SQL test harness and its test coverage.
### How was this patch tested?
The following focused tests pass:
```
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt \
'sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z "test
triggered flow"'
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt \
'sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z
"streaming/available-now"'
```
The second command runs five execution golden tests and their five analyzer
tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #59665 from siying/codex/sql-golden-streaming-flows.
Authored-by: Siying Dong <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 2d4a7c59b5f473618c432d7712767156bf53fafc)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../streaming/available-now-multiple-flows.sql.out | 191 ++++++
.../available-now-stream-static-join.sql.out | 113 ++++
.../streaming/available-now-to-table.sql.out | 99 +++
.../streaming/available-now-union.sql.out | 118 ++++
.../available-now-window-aggregation.sql.out | 120 ++++
.../streaming/available-now-multiple-flows.sql | 48 ++
.../streaming/available-now-stream-static-join.sql | 31 +
.../inputs/streaming/available-now-to-table.sql | 27 +
.../inputs/streaming/available-now-union.sql | 31 +
.../streaming/available-now-window-aggregation.sql | 46 ++
.../streaming/available-now-multiple-flows.sql.out | 214 ++++++
.../available-now-stream-static-join.sql.out | 123 ++++
.../streaming/available-now-to-table.sql.out | 111 +++
.../results/streaming/available-now-union.sql.out | 124 ++++
.../available-now-window-aggregation.sql.out | 128 ++++
.../org/apache/spark/sql/SQLQueryTestHelper.scala | 8 +
.../org/apache/spark/sql/SQLQueryTestSuite.scala | 64 +-
.../spark/sql/StreamingSQLQueryTestSuite.scala | 745 +++++++++++++++++++++
18 files changed, 2328 insertions(+), 13 deletions(-)
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-multiple-flows.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-multiple-flows.sql.out
new file mode 100644
index 000000000000..67269e8a29f6
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-multiple-flows.sql.out
@@ -0,0 +1,191 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_source
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_source
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_intermediate
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_intermediate
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_sink
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_sink
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_source
+ (id INT, value STRING) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_multiple_flows_source`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_multiple_flows_source
+VALUES (1, 'one'), (2, 'two')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source],
Append, `spark_catalog`.`default`.`streaming_sql_golden_multiple_flows_source`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source),
[id, value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_intermediate
+ (id INT, value STRING) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_multiple_flows_intermediate`,
false
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_sink
+ (id INT, value STRING) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_multiple_flows_sink`, false
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_multiple_flows_downstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_sink BY NAME
+SELECT id, concat(value, '!') AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_intermediate)
+-- !query analysis
+~Project [id#x, concat(value#x, !) AS value#x]
++- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_intermediate
+ +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_intermediate],
[id#x, value#x], name=<Unassigned>
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_multiple_flows_upstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_intermediate BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_source)
+-- !query analysis
+~Project [id#x, upper(value#x) AS value#x]
++- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_source
+ +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source], [id#x,
value#x], name=<Unassigned>
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_sink[id#x,value#x]
parquet
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_intermediate ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_intermediate
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_intermediate[id#x,value#x]
parquet
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_sink[id#x,value#x]
parquet
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_sink[id#x,value#x]
parquet
+
+
+-- !query
+INSERT INTO streaming_sql_golden_multiple_flows_source VALUES (3, 'three')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source],
Append, `spark_catalog`.`default`.`streaming_sql_golden_multiple_flows_source`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_multiple_flows_source),
[id, value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_sink[id#x,value#x]
parquet
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream
+-- !query analysis
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_multiple_flows_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_multiple_flows_sink[id#x,value#x]
parquet
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_source
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_source
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_intermediate
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_intermediate
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_sink
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_multiple_flows_sink
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-stream-static-join.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-stream-static-join.sql.out
new file mode 100644
index 000000000000..22e29024268d
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-stream-static-join.sql.out
@@ -0,0 +1,113 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_events
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_events
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_customers
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_customers
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_sink
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_sink
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_events (
+ event_id INT, customer_id INT, amount INT) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_join_events`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_join_events
+VALUES (1, 10, 25), (2, 20, 40), (3, 99, 15)
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_join_events, false, Parquet,
[path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_join_events], Append,
`spark_catalog`.`default`.`streaming_sql_golden_join_events`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_join_events), [event_id,
customer_id, amount]
++- Project [col1#x AS event_id#x, col2#x AS customer_id#x, col3#x AS amount#x]
+ +- LocalRelation [col1#x, col2#x, col3#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_customers (
+ customer_id INT, tier STRING) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_join_customers`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_join_customers
+VALUES (10, 'gold'), (20, 'silver')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_join_customers, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_join_customers], Append,
`spark_catalog`.`default`.`streaming_sql_golden_join_customers`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_join_customers),
[customer_id, tier]
++- Project [col1#x AS customer_id#x, col2#x AS tier#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_sink (
+ event_id INT, tier STRING, amount INT) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_join_sink`, false
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_join_flow AS
+INSERT INTO streaming_sql_golden_join_sink BY NAME
+SELECT events.event_id, coalesce(customers.tier, 'unknown') AS tier,
events.amount
+FROM STREAM(streaming_sql_golden_join_events) AS events
+LEFT JOIN streaming_sql_golden_join_customers AS customers
+ ON events.customer_id = customers.customer_id
+-- !query analysis
+~Project [event_id#x, coalesce(tier#x, unknown) AS tier#x, amount#x]
++- ~Join LeftOuter, (customer_id#x = customer_id#x)
+ :- ~SubqueryAlias events
+ : +- ~SubqueryAlias spark_catalog.default.streaming_sql_golden_join_events
+ : +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_join_events], [event_id#x,
customer_id#x, amount#x], name=<Unassigned>
+ +- SubqueryAlias customers
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_join_customers
+ +- Relation
spark_catalog.default.streaming_sql_golden_join_customers[customer_id#x,tier#x]
parquet
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_join_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_join_sink ORDER BY event_id
+-- !query analysis
+Sort [event_id#x ASC NULLS FIRST], true
++- Project [event_id#x, tier#x, amount#x]
+ +- SubqueryAlias spark_catalog.default.streaming_sql_golden_join_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_join_sink[event_id#x,tier#x,amount#x]
parquet
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_events
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_events
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_customers
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_customers
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_sink
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_join_sink
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-to-table.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-to-table.sql.out
new file mode 100644
index 000000000000..b61e35e80a25
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-to-table.sql.out
@@ -0,0 +1,99 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_source
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_available_now_source
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_sink
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_available_now_sink
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_available_now_source (id INT, value STRING)
USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_available_now_source`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_available_now_source
+VALUES (1, 'one'), (2, 'two'), (3, 'three')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source], Append,
`spark_catalog`.`default`.`streaming_sql_golden_available_now_source`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source), [id,
value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_available_now_sink (id INT, value STRING)
USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_available_now_sink`, false
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_available_now_flow AS
+INSERT INTO streaming_sql_golden_available_now_sink BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_available_now_source)
+WHERE id >= 2
+-- !query analysis
+~Project [id#x, upper(value#x) AS value#x]
++- ~Filter (id#x >= 2)
+ +- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_available_now_source
+ +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source], [id#x,
value#x], name=<Unassigned>
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_available_now_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_available_now_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_available_now_sink[id#x,value#x]
parquet
+
+
+-- !query
+INSERT INTO streaming_sql_golden_available_now_source VALUES (4, 'four')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source], Append,
`spark_catalog`.`default`.`streaming_sql_golden_available_now_source`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_available_now_source), [id,
value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_available_now_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, value#x]
+ +- SubqueryAlias
spark_catalog.default.streaming_sql_golden_available_now_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_available_now_sink[id#x,value#x]
parquet
+
+
+-- !query
+DROP TABLE streaming_sql_golden_available_now_source
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_available_now_source
+
+
+-- !query
+DROP TABLE streaming_sql_golden_available_now_sink
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_available_now_sink
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-union.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-union.sql.out
new file mode 100644
index 000000000000..5708eb8b854c
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-union.sql.out
@@ -0,0 +1,118 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_left
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_left
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_right
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_right
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_sink
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_sink
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_left (id INT, value STRING) USING
parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_union_left`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_union_left VALUES (1, 'alpha'), (2, 'beta')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_left, false, Parquet,
[path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_left], Append,
`spark_catalog`.`default`.`streaming_sql_golden_union_left`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_union_left), [id, value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_right (id INT, value STRING) USING
parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_union_right`, false
+
+
+-- !query
+INSERT INTO streaming_sql_golden_union_right VALUES (3, 'gamma'), (4, 'delta')
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_right, false, Parquet,
[path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_right], Append,
`spark_catalog`.`default`.`streaming_sql_golden_union_right`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_union_right), [id, value]
++- Project [col1#x AS id#x, col2#x AS value#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_sink (
+ id INT, origin STRING, value STRING) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_union_sink`, false
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_union_flow AS
+INSERT INTO streaming_sql_golden_union_sink BY NAME
+WITH normalized AS (
+ SELECT id, 'left' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_left)
+ UNION ALL
+ SELECT id, 'right' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_right)
+)
+SELECT id, origin, value FROM normalized
+-- !query analysis
+~WithCTE
+:- ~CTERelationDef xxxx, false
+: +- ~SubqueryAlias normalized
+: +- ~Union false, false
+: :- ~Project [id#x, left AS origin#x, upper(value#x) AS value#x]
+: : +- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_union_left
+: : +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_left], [id#x, value#x],
name=<Unassigned>
+: +- ~Project [id#x, right AS origin#x, upper(value#x) AS value#x]
+: +- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_union_right
+: +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_union_right], [id#x, value#x],
name=<Unassigned>
++- ~Project [id#x, origin#x, value#x]
+ +- ~SubqueryAlias normalized
+ +- ~CTERelationRef xxxx, true, [id#x, origin#x, value#x], true, false
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_union_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_union_sink ORDER BY id
+-- !query analysis
+Sort [id#x ASC NULLS FIRST], true
++- Project [id#x, origin#x, value#x]
+ +- SubqueryAlias spark_catalog.default.streaming_sql_golden_union_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_union_sink[id#x,origin#x,value#x]
parquet
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_left
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_left
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_right
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_right
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_sink
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_union_sink
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-window-aggregation.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-window-aggregation.sql.out
new file mode 100644
index 000000000000..4b7977f975b4
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-window-aggregation.sql.out
@@ -0,0 +1,120 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_events
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_aggregation_events
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_sink
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_aggregation_sink
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_aggregation_events (
+ event_time TIMESTAMP, category STRING, amount INT) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_aggregation_events`, false
+
+
+-- !query
+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)
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events], Append,
`spark_catalog`.`default`.`streaming_sql_golden_aggregation_events`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events),
[event_time, category, amount]
++- Project [col1#x AS event_time#x, col2#x AS category#x, col3#x AS amount#x]
+ +- LocalRelation [col1#x, col2#x, col3#x]
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_aggregation_sink (
+ window_start TIMESTAMP,
+ window_end TIMESTAMP,
+ category STRING,
+ total BIGINT,
+ event_count BIGINT) USING parquet
+-- !query analysis
+CreateDataSourceTableCommand
`spark_catalog`.`default`.`streaming_sql_golden_aggregation_sink`, false
+
+
+-- !query
+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
+-- !query analysis
+~Aggregate [window#x-T10000ms, category#x], [window#x-T10000ms.start AS
window_start#x, window#x-T10000ms.end AS window_end#x, category#x,
sum(amount#x) AS total#xL, count(1) AS event_count#xL]
++- ~Project [named_struct(start,
knownnullable(precisetimestampconversion(((precisetimestampconversion(event_time#x-T10000ms,
TimestampType, LongType) - CASE WHEN
(((precisetimestampconversion(event_time#x-T10000ms, TimestampType, LongType) -
0) % 60000000) < cast(0 as bigint)) THEN
(((precisetimestampconversion(event_time#x-T10000ms, TimestampType, LongType) -
0) % 60000000) + 60000000) ELSE
((precisetimestampconversion(event_time#x-T10000ms, TimestampType, LongType) -
0) % 60000000) EN [...]
+ +- ~Filter isnotnull(event_time#x-T10000ms)
+ +- ~EventTimeWatermark xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx,
event_time#x: timestamp, 10 seconds
+ +- ~SubqueryAlias
spark_catalog.default.streaming_sql_golden_aggregation_events
+ +- StreamingRelation FileSource[file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events],
[event_time#x, category#x, amount#x], name=<Unassigned>
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_aggregation_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category
+-- !query analysis
+Sort [window_start#x ASC NULLS FIRST, category#x ASC NULLS FIRST], true
++- Project [window_start#x, window_end#x, category#x, total#xL, event_count#xL]
+ +- SubqueryAlias spark_catalog.default.streaming_sql_golden_aggregation_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_aggregation_sink[window_start#x,window_end#x,category#x,total#xL,event_count#xL]
parquet
+
+
+-- !query
+INSERT INTO streaming_sql_golden_aggregation_events VALUES
+ (TIMESTAMP '2024-01-01 00:10:00', '__watermark__', 0)
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events, false,
Parquet, [path=file:[not included in
comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events], Append,
`spark_catalog`.`default`.`streaming_sql_golden_aggregation_events`,
org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included
in comparison]/{warehouse_dir}/streaming_sql_golden_aggregation_events),
[event_time, category, amount]
++- Project [col1#x AS event_time#x, col2#x AS category#x, col3#x AS amount#x]
+ +- LocalRelation [col1#x, col2#x, col3#x]
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_aggregation_flow
+-- !query analysis
+
+
+
+-- !query
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category
+-- !query analysis
+Sort [window_start#x ASC NULLS FIRST, category#x ASC NULLS FIRST], true
++- Project [window_start#x, window_end#x, category#x, total#xL, event_count#xL]
+ +- SubqueryAlias spark_catalog.default.streaming_sql_golden_aggregation_sink
+ +- Relation
spark_catalog.default.streaming_sql_golden_aggregation_sink[window_start#x,window_end#x,category#x,total#xL,event_count#xL]
parquet
+
+
+-- !query
+DROP TABLE streaming_sql_golden_aggregation_events
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_aggregation_events
+
+
+-- !query
+DROP TABLE streaming_sql_golden_aggregation_sink
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog),
default.streaming_sql_golden_aggregation_sink
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-multiple-flows.sql
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-multiple-flows.sql
new file mode 100644
index 000000000000..b89ea6cee5a2
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-multiple-flows.sql
@@ -0,0 +1,48 @@
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_source;
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_intermediate;
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_sink;
+
+CREATE TABLE streaming_sql_golden_multiple_flows_source
+ (id INT, value STRING) USING parquet;
+INSERT INTO streaming_sql_golden_multiple_flows_source
+VALUES (1, 'one'), (2, 'two');
+
+CREATE TABLE streaming_sql_golden_multiple_flows_intermediate
+ (id INT, value STRING) USING parquet;
+CREATE TABLE streaming_sql_golden_multiple_flows_sink
+ (id INT, value STRING) USING parquet;
+
+CREATE FLOW streaming_sql_golden_multiple_flows_downstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_sink BY NAME
+SELECT id, concat(value, '!') AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_intermediate);
+
+CREATE FLOW streaming_sql_golden_multiple_flows_upstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_intermediate BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_source);
+
+-- Start the downstream flow before its upstream flow has produced any rows.
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream;
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id;
+
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream;
+SELECT * FROM streaming_sql_golden_multiple_flows_intermediate ORDER BY id;
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id;
+
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream;
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id;
+
+INSERT INTO streaming_sql_golden_multiple_flows_source VALUES (3, 'three');
+
+-- Interleave the refreshes in the opposite order. The downstream flow has no
new input yet.
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream;
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id;
+
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream;
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream;
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id;
+
+DROP TABLE streaming_sql_golden_multiple_flows_source;
+DROP TABLE streaming_sql_golden_multiple_flows_intermediate;
+DROP TABLE streaming_sql_golden_multiple_flows_sink;
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-stream-static-join.sql
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-stream-static-join.sql
new file mode 100644
index 000000000000..f16956c2500c
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-stream-static-join.sql
@@ -0,0 +1,31 @@
+DROP TABLE IF EXISTS streaming_sql_golden_join_events;
+DROP TABLE IF EXISTS streaming_sql_golden_join_customers;
+DROP TABLE IF EXISTS streaming_sql_golden_join_sink;
+
+CREATE TABLE streaming_sql_golden_join_events (
+ event_id INT, customer_id INT, amount INT) USING parquet;
+INSERT INTO streaming_sql_golden_join_events
+VALUES (1, 10, 25), (2, 20, 40), (3, 99, 15);
+
+CREATE TABLE streaming_sql_golden_join_customers (
+ customer_id INT, tier STRING) USING parquet;
+INSERT INTO streaming_sql_golden_join_customers
+VALUES (10, 'gold'), (20, 'silver');
+
+CREATE TABLE streaming_sql_golden_join_sink (
+ event_id INT, tier STRING, amount INT) USING parquet;
+
+CREATE FLOW streaming_sql_golden_join_flow AS
+INSERT INTO streaming_sql_golden_join_sink BY NAME
+SELECT events.event_id, coalesce(customers.tier, 'unknown') AS tier,
events.amount
+FROM STREAM(streaming_sql_golden_join_events) AS events
+LEFT JOIN streaming_sql_golden_join_customers AS customers
+ ON events.customer_id = customers.customer_id;
+
+REFRESH FLOW streaming_sql_golden_join_flow;
+
+SELECT * FROM streaming_sql_golden_join_sink ORDER BY event_id;
+
+DROP TABLE streaming_sql_golden_join_events;
+DROP TABLE streaming_sql_golden_join_customers;
+DROP TABLE streaming_sql_golden_join_sink;
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-to-table.sql
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-to-table.sql
new file mode 100644
index 000000000000..e70d53f6d283
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-to-table.sql
@@ -0,0 +1,27 @@
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_source;
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_sink;
+
+CREATE TABLE streaming_sql_golden_available_now_source (id INT, value STRING)
USING parquet;
+INSERT INTO streaming_sql_golden_available_now_source
+VALUES (1, 'one'), (2, 'two'), (3, 'three');
+
+CREATE TABLE streaming_sql_golden_available_now_sink (id INT, value STRING)
USING parquet;
+
+CREATE FLOW streaming_sql_golden_available_now_flow AS
+INSERT INTO streaming_sql_golden_available_now_sink BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_available_now_source)
+WHERE id >= 2;
+
+REFRESH FLOW streaming_sql_golden_available_now_flow;
+
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id;
+
+INSERT INTO streaming_sql_golden_available_now_source VALUES (4, 'four');
+
+REFRESH FLOW streaming_sql_golden_available_now_flow;
+
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id;
+
+DROP TABLE streaming_sql_golden_available_now_source;
+DROP TABLE streaming_sql_golden_available_now_sink;
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-union.sql
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-union.sql
new file mode 100644
index 000000000000..c016210499ce
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-union.sql
@@ -0,0 +1,31 @@
+DROP TABLE IF EXISTS streaming_sql_golden_union_left;
+DROP TABLE IF EXISTS streaming_sql_golden_union_right;
+DROP TABLE IF EXISTS streaming_sql_golden_union_sink;
+
+CREATE TABLE streaming_sql_golden_union_left (id INT, value STRING) USING
parquet;
+INSERT INTO streaming_sql_golden_union_left VALUES (1, 'alpha'), (2, 'beta');
+
+CREATE TABLE streaming_sql_golden_union_right (id INT, value STRING) USING
parquet;
+INSERT INTO streaming_sql_golden_union_right VALUES (3, 'gamma'), (4, 'delta');
+
+CREATE TABLE streaming_sql_golden_union_sink (
+ id INT, origin STRING, value STRING) USING parquet;
+
+CREATE FLOW streaming_sql_golden_union_flow AS
+INSERT INTO streaming_sql_golden_union_sink BY NAME
+WITH normalized AS (
+ SELECT id, 'left' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_left)
+ UNION ALL
+ SELECT id, 'right' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_right)
+)
+SELECT id, origin, value FROM normalized;
+
+REFRESH FLOW streaming_sql_golden_union_flow;
+
+SELECT * FROM streaming_sql_golden_union_sink ORDER BY id;
+
+DROP TABLE streaming_sql_golden_union_left;
+DROP TABLE streaming_sql_golden_union_right;
+DROP TABLE streaming_sql_golden_union_sink;
diff --git
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-window-aggregation.sql
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-window-aggregation.sql
new file mode 100644
index 000000000000..9937e4c83303
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-window-aggregation.sql
@@ -0,0 +1,46 @@
+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;
+
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category;
+
+INSERT INTO streaming_sql_golden_aggregation_events VALUES
+ (TIMESTAMP '2024-01-01 00:10:00', '__watermark__', 0);
+
+REFRESH FLOW streaming_sql_golden_aggregation_flow;
+
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category;
+
+DROP TABLE streaming_sql_golden_aggregation_events;
+DROP TABLE streaming_sql_golden_aggregation_sink;
diff --git
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-multiple-flows.sql.out
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-multiple-flows.sql.out
new file mode 100644
index 000000000000..e13458c28b81
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-multiple-flows.sql.out
@@ -0,0 +1,214 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_source
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_intermediate
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_multiple_flows_sink
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_source
+ (id INT, value STRING) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_multiple_flows_source
+VALUES (1, 'one'), (2, 'two')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_intermediate
+ (id INT, value STRING) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_multiple_flows_sink
+ (id INT, value STRING) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_multiple_flows_downstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_sink BY NAME
+SELECT id, concat(value, '!') AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_intermediate)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_multiple_flows_upstream AS
+INSERT INTO streaming_sql_golden_multiple_flows_intermediate BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_multiple_flows_source)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_intermediate ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+1 ONE
+2 TWO
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+1 ONE!
+2 TWO!
+
+
+-- !query
+INSERT INTO streaming_sql_golden_multiple_flows_source VALUES (3, 'three')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+1 ONE!
+2 TWO!
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_upstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_multiple_flows_downstream
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_multiple_flows_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+1 ONE!
+2 TWO!
+3 THREE!
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_source
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_intermediate
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_multiple_flows_sink
+-- !query schema
+struct<>
+-- !query output
+
diff --git
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-stream-static-join.sql.out
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-stream-static-join.sql.out
new file mode 100644
index 000000000000..5e6218b0abc1
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-stream-static-join.sql.out
@@ -0,0 +1,123 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_events
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_customers
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_join_sink
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_events (
+ event_id INT, customer_id INT, amount INT) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_join_events
+VALUES (1, 10, 25), (2, 20, 40), (3, 99, 15)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_customers (
+ customer_id INT, tier STRING) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_join_customers
+VALUES (10, 'gold'), (20, 'silver')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_join_sink (
+ event_id INT, tier STRING, amount INT) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_join_flow AS
+INSERT INTO streaming_sql_golden_join_sink BY NAME
+SELECT events.event_id, coalesce(customers.tier, 'unknown') AS tier,
events.amount
+FROM STREAM(streaming_sql_golden_join_events) AS events
+LEFT JOIN streaming_sql_golden_join_customers AS customers
+ ON events.customer_id = customers.customer_id
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_join_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_join_sink ORDER BY event_id
+-- !query schema
+struct<event_id:int,tier:string,amount:int>
+-- !query output
+1 gold 25
+2 silver 40
+3 unknown 15
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_events
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_customers
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_join_sink
+-- !query schema
+struct<>
+-- !query output
+
diff --git
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-to-table.sql.out
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-to-table.sql.out
new file mode 100644
index 000000000000..1775d3758e56
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-to-table.sql.out
@@ -0,0 +1,111 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_source
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_available_now_sink
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_available_now_source (id INT, value STRING)
USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_available_now_source
+VALUES (1, 'one'), (2, 'two'), (3, 'three')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_available_now_sink (id INT, value STRING)
USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_available_now_flow AS
+INSERT INTO streaming_sql_golden_available_now_sink BY NAME
+SELECT id, upper(value) AS value
+FROM STREAM(streaming_sql_golden_available_now_source)
+WHERE id >= 2
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_available_now_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+2 TWO
+3 THREE
+
+
+-- !query
+INSERT INTO streaming_sql_golden_available_now_source VALUES (4, 'four')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_available_now_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_available_now_sink ORDER BY id
+-- !query schema
+struct<id:int,value:string>
+-- !query output
+2 TWO
+3 THREE
+4 FOUR
+
+
+-- !query
+DROP TABLE streaming_sql_golden_available_now_source
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_available_now_sink
+-- !query schema
+struct<>
+-- !query output
+
diff --git
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-union.sql.out
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-union.sql.out
new file mode 100644
index 000000000000..9c2e081d4af4
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-union.sql.out
@@ -0,0 +1,124 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_left
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_right
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_union_sink
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_left (id INT, value STRING) USING
parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_union_left VALUES (1, 'alpha'), (2, 'beta')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_right (id INT, value STRING) USING
parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_union_right VALUES (3, 'gamma'), (4, 'delta')
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_union_sink (
+ id INT, origin STRING, value STRING) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE FLOW streaming_sql_golden_union_flow AS
+INSERT INTO streaming_sql_golden_union_sink BY NAME
+WITH normalized AS (
+ SELECT id, 'left' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_left)
+ UNION ALL
+ SELECT id, 'right' AS origin, upper(value) AS value
+ FROM STREAM(streaming_sql_golden_union_right)
+)
+SELECT id, origin, value FROM normalized
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_union_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM streaming_sql_golden_union_sink ORDER BY id
+-- !query schema
+struct<id:int,origin:string,value:string>
+-- !query output
+1 left ALPHA
+2 left BETA
+3 right GAMMA
+4 right DELTA
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_left
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_right
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_union_sink
+-- !query schema
+struct<>
+-- !query output
+
diff --git
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-window-aggregation.sql.out
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-window-aggregation.sql.out
new file mode 100644
index 000000000000..8ceb34557bb1
--- /dev/null
+++
b/sql/core/src/test/resources/sql-tests/results/streaming/available-now-window-aggregation.sql.out
@@ -0,0 +1,128 @@
+-- Automatically generated by StreamingSQLQueryTestSuite
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_events
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE IF EXISTS streaming_sql_golden_aggregation_sink
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_aggregation_events (
+ event_time TIMESTAMP, category STRING, amount INT) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+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)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+CREATE TABLE streaming_sql_golden_aggregation_sink (
+ window_start TIMESTAMP,
+ window_end TIMESTAMP,
+ category STRING,
+ total BIGINT,
+ event_count BIGINT) USING parquet
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+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
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_aggregation_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category
+-- !query schema
+struct<window_start:timestamp,window_end:timestamp,category:string,total:bigint,event_count:bigint>
+-- !query output
+
+
+
+-- !query
+INSERT INTO streaming_sql_golden_aggregation_events VALUES
+ (TIMESTAMP '2024-01-01 00:10:00', '__watermark__', 0)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+REFRESH FLOW streaming_sql_golden_aggregation_flow
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT *
+FROM streaming_sql_golden_aggregation_sink
+ORDER BY window_start, category
+-- !query schema
+struct<window_start:timestamp,window_end:timestamp,category:string,total:bigint,event_count:bigint>
+-- !query output
+2024-01-01 00:00:00 2024-01-01 00:01:00 books 5 2
+2024-01-01 00:01:00 2024-01-01 00:02:00 games 7 1
+
+
+-- !query
+DROP TABLE streaming_sql_golden_aggregation_events
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+DROP TABLE streaming_sql_golden_aggregation_sink
+-- !query schema
+struct<>
+-- !query output
+
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestHelper.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestHelper.scala
index e4782b7fd41e..31e7281d31f5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestHelper.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestHelper.scala
@@ -51,6 +51,10 @@ trait SQLQueryTestHelper extends SQLConfHelper with Logging {
line.replaceAll("#\\d+", "#x")
.replaceAll("plan_id=\\d+", "plan_id=x")
.replaceAll("uuid\\(Some\\(-?\\d+\\)\\)", "uuid(Some(x))")
+ .replaceAll(
+ "(?i)(EventTimeWatermark )\\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-" +
+ "[0-9a-f]{4}-[0-9a-f]{12}\\b",
+ "$1xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")
.replaceAll(
s"Location.*$clsName/",
s"Location $notIncludedMsg/{warehouse_dir}/")
@@ -93,6 +97,10 @@ trait SQLQueryTestHelper extends SQLConfHelper with Logging {
// objects such as views. Therefore any following queries that reference
these objects should
// find them in the catalog.
val df = session.sql(sql)
+ getNormalizedQueryAnalysisResult(df)
+ }
+
+ protected def getNormalizedQueryAnalysisResult(df: DataFrame): (String,
Seq[String]) = {
val schema = df.schema.catalogString
val analyzed = df.queryExecution.analyzed
// Determine if the analyzed plan contains any nondeterministic
expressions.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
index a0b481e1a56c..03188118bcce 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
@@ -64,6 +64,12 @@ import org.apache.spark.util.Utils
* SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly
org.apache.spark.sql.SQLQueryTestSuite -- -z describe.sql"
* }}}
*
+ * The commands above exclude inputs under "streaming/". To re-generate their
execution and
+ * analyzer golden files, run:
+ * {{{
+ * SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly
org.apache.spark.sql.StreamingSQLQueryTestSuite"
+ * }}}
+ *
* The format for input files is simple:
* 1. A list of SQL queries separated by semicolons by default. If the
semicolon cannot effectively
* separate the SQL queries in the test file(e.g. bracketed comments),
please use
@@ -187,7 +193,8 @@ class SQLQueryTestSuite extends SharedSparkSession
if (TestUtils.testCommandAvailable("/bin/bash")) Nil else
Set("transform.sql")
/** List of test cases to ignore, in lower cases. */
protected def ignoreList: Set[String] = Set(
- "ignored.sql" // Do NOT remove this one. It is here to test the ignore
functionality.
+ "ignored.sql", // Do NOT remove this one. It is here to test the ignore
functionality.
+ "streaming/" // Handled by StreamingSQLQueryTestSuite.
) ++ otherIgnoreList
/** List of test cases that require TPCDS table schemas to be loaded. */
private def requireTPCDSCases: Seq[String] = Seq("pipe-operators.sql")
@@ -366,6 +373,27 @@ class SQLQueryTestSuite extends SharedSparkSession
localSparkSession
}
+ protected def parsePlan(session: SparkSession, sql: String):
Option[LogicalPlan] = {
+ session.withActive {
+ try {
+ Some(session.sessionState.sqlParser.parsePlan(sql))
+ } catch {
+ case _: ParseException => None
+ }
+ }
+ }
+
+ /**
+ * Returns a per-test executor for SQL statements handled by a specialized
subclass. The plan
+ * supplier is lazy so ordinary statements are not parsed an extra time.
+ */
+ protected def createCustomQueryExecutor(
+ session: SparkSession,
+ testCase: TestCase,
+ sparkConfigSet: Seq[(String, String)],
+ analyzerOnly: Boolean):
+ (String, () => Option[LogicalPlan]) => Option[(String, Seq[String])] =
(_, _) => None
+
/**
* Runs queries in debug mode. Only commands and --DEBUG-marked queries are
executed.
* Failed debug queries print the full error stacktrace. Results are
compared against
@@ -382,6 +410,8 @@ class SQLQueryTestSuite extends SharedSparkSession
val localSparkSession = setupLocalSession(testCase, sparkConfigSet)
val lowercaseTestCase = testCase.name.toLowerCase(Locale.ROOT)
val segmentsPerQuery = 3
+ val executeCustomQuery =
+ createCustomQueryExecutor(localSparkSession, testCase, sparkConfigSet,
analyzerOnly = false)
val goldenFileExists = new File(testCase.resultFile).exists()
val segments = if (goldenFileExists) {
@@ -391,15 +421,13 @@ class SQLQueryTestSuite extends SharedSparkSession
}
queriesWithDebug.zipWithIndex.foreach { case ((sql, isDebug), i) =>
- val isCommand = try {
-
localSparkSession.sessionState.sqlParser.parsePlan(sql).isInstanceOf[Command]
- } catch {
- case _: ParseException => false
- }
+ lazy val parsedPlan = parsePlan(localSparkSession, sql)
+ val customQueryExecuted = executeCustomQuery(sql, () =>
parsedPlan).nonEmpty
+ lazy val isCommand = parsedPlan.exists(_.isInstanceOf[Command])
- if (isCommand) {
+ if (!customQueryExecuted && isCommand) {
localSparkSession.sql(sql).collect()
- } else if (isDebug) {
+ } else if (!customQueryExecuted && isDebug) {
// Capture exception stacktrace if the query fails.
var exceptionTrace: Option[String] = None
val (schema, result) = handleExceptions {
@@ -483,21 +511,29 @@ class SQLQueryTestSuite extends SharedSparkSession
// This does not isolate catalog changes.
val localSparkSession = setupLocalSession(testCase, sparkConfigSet)
val lowercaseTestCase = testCase.name.toLowerCase(Locale.ROOT)
+ val analyzerOnly = testCase.isInstanceOf[AnalyzerTest]
+ val executeCustomQuery =
+ createCustomQueryExecutor(localSparkSession, testCase, sparkConfigSet,
analyzerOnly)
// Run the SQL queries preparing them for comparison.
val outputs: Seq[QueryTestOutput] = queries.map { sql =>
+ lazy val parsedPlan = parsePlan(localSparkSession, sql)
testCase match {
case _: AnalyzerTest =>
- val (_, output) =
-
handleExceptions(getNormalizedQueryAnalysisResult(localSparkSession, sql))
+ val (_, output) = handleExceptions {
+ executeCustomQuery(sql, () => parsedPlan)
+ .getOrElse(getNormalizedQueryAnalysisResult(localSparkSession,
sql))
+ }
// We do some query canonicalization now.
AnalyzerOutput(
sql = sql,
schema = None,
output = normalizeTestResults(output.mkString("\n")))
case _ =>
- val (schema, output) =
-
handleExceptions(getNormalizedQueryExecutionResult(localSparkSession, sql))
+ val (schema, output) = handleExceptions {
+ executeCustomQuery(sql, () => parsedPlan)
+ .getOrElse(getNormalizedQueryExecutionResult(localSparkSession,
sql))
+ }
// We do some query canonicalization now.
val executionOutput = ExecutionOutput(
sql = sql,
@@ -576,7 +612,9 @@ class SQLQueryTestSuite extends SharedSparkSession
file.getAbsolutePath.replace(inputFilePath, goldenFilePath) + ".out"
}
- protected lazy val listTestCases: Seq[TestCase] = {
+ protected lazy val listTestCases: Seq[TestCase] = createTestCases()
+
+ protected def createTestCases(): Seq[TestCase] = {
listFilesRecursively(new File(inputFilePath)).flatMap { file =>
var resultFile = resultFileForInputFile(file)
var analyzerResultFile =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala
new file mode 100644
index 000000000000..d0197e330be3
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala
@@ -0,0 +1,745 @@
+/*
+ * 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"
+ * }}}
+ *
+ * To re-generate all execution and analyzer golden files for streaming
inputs, run:
+ * {{{
+ * SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly
org.apache.spark.sql.StreamingSQLQueryTestSuite"
+ * }}}
+ *
+ * To re-generate both golden files for one streaming input, run:
+ * {{{
+ * SPARK_GENERATE_GOLDEN_FILES=1 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 ResolvedTableName(
+ catalog: String,
+ namespace: Seq[String],
+ table: String) {
+ def sql: String = (catalog +: namespace :+
table).map(quoteIdentifier).mkString(".")
+ }
+
+ private case class TriggeredFlow(
+ flowName: Seq[String],
+ query: LogicalPlan,
+ targetTable: String,
+ resolvedTargetTable: ResolvedTableName,
+ checkpointLocation: 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 resolveTableName(
+ session: SparkSession,
+ multipartIdentifier: Seq[String]): ResolvedTableName = {
+ val analyzer = session.sessionState.analyzer
+ import analyzer.CatalogAndIdentifier
+
+ val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier
+ ResolvedTableName(catalog.name(), identifier.namespace().toSeq,
identifier.name())
+ }
+
+ private def sameTableName(
+ session: SparkSession,
+ left: ResolvedTableName,
+ right: ResolvedTableName): Boolean = {
+ val resolver = session.sessionState.conf.resolver
+ resolver(left.catalog, right.catalog) &&
+ left.namespace.length == right.namespace.length &&
+ left.namespace.zip(right.namespace).forall { case (l, r) => resolver(l,
r) } &&
+ resolver(left.table, right.table)
+ }
+
+ private def parseCreateTriggeredFlow(
+ session: SparkSession,
+ 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, resolvedTargetTable) = insert.table match {
+ case target: UnresolvedInsertTarget =>
+ val targetTable =
target.multipartIdentifier.map(quoteIdentifier).mkString(".")
+ targetTable -> resolveTableName(session, target.multipartIdentifier)
+ case _ =>
+ throw new IllegalArgumentException(
+ "SQL golden triggered flows require a static table identifier as the
INSERT target.")
+ }
+ val checkpointLocation = new Path(
+ session.sessionState.conf.warehousePath,
+ s"sql-query-test-checkpoint-${UUID.randomUUID()}").toString
+ TriggeredFlow(
+ flowName,
+ insert.query,
+ targetTable,
+ resolvedTargetTable,
+ checkpointLocation)
+ }
+
+ private def registerTriggeredFlow(
+ session: SparkSession,
+ command: CreateFlowCommand,
+ flows: mutable.Map[Seq[String], TriggeredFlow]): TriggeredFlow = {
+ session.withActive {
+ val flow = parseCreateTriggeredFlow(session, 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 =>
+ sameTableName(session, registeredFlow.resolvedTargetTable,
flow.resolvedTargetTable)
+ }.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): RefreshTriggeredFlow = {
+ try {
+
RefreshTriggeredFlow(session.sessionState.sqlParser.parseMultipartIdentifier(name))
+ } catch {
+ case e: ParseException =>
+ throw new IllegalArgumentException(
+ s"REFRESH FLOW requires a valid flow name, but '$name' is invalid.",
e)
+ }
+ }
+
+ private def parseTriggeredFlowStatement(
+ session: SparkSession,
+ sql: String,
+ parsedPlan: => Option[LogicalPlan]): Option[TriggeredFlowStatement] = {
+ session.withActive {
+ val tokens = sql.trim.split("\\s+", 3)
+ if (tokens.length >= 2 && tokens(1).equalsIgnoreCase("FLOW")) {
+ if (tokens(0).equalsIgnoreCase("REFRESH")) {
+ if (tokens.length == 2) {
+ throw new IllegalArgumentException("REFRESH FLOW requires a flow
name.")
+ }
+ Some(parseRefreshTriggeredFlowStatement(session, tokens(2)))
+ } else if (tokens(0).equalsIgnoreCase("CREATE")) {
+ parsedPlan.collect {
+ case command: CreateFlowCommand => CreateTriggeredFlow(command)
+ }
+ } else {
+ 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 targetTableName = flow.resolvedTargetTable.sql
+ 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(targetTableName),
+ s"SQL golden triggered flow target table '${flow.targetTable}' does
not exist.")
+ val queryName = triggeredFlowQueryName(testCase, sparkConfigSet,
flow.flowName)
+ val targetTable = loadV1Table(session, targetTableName)
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
+ targetTableName,
+ session.table(targetTableName).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", flow.checkpointLocation)
+ targetTable.foreach { table =>
+ writer.format(table.provider.getOrElse {
+ throw new IllegalArgumentException(
+ s"SQL golden triggered flow target table '${flow.targetTable}' has
no provider.")
+ })
+ writer.options(table.storage.properties)
+ if (table.partitionColumnNames.nonEmpty) {
+ writer.partitionBy(table.partitionColumnNames: _*)
+ }
+ }
+ val query = writer.toTable(targetTableName)
+ 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 `spark_catalog`.`default`.`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 `spark_catalog`.`default`.`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 == "`spark_catalog`.`default`.`sink-table`")
+ assert(resolveTriggeredFlow(spark, flowName, flows) eq flow)
+ val separatelyRegisteredFlow = registerTriggeredFlow(
+ spark,
+ createFlowCommand(specs(1)._1),
+ mutable.Map.empty)
+ assert(separatelyRegisteredFlow.checkpointLocation !=
flow.checkpointLocation)
+ 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]))
+ val refreshWithoutParsedPlan = parseTriggeredFlowStatement(
+ spark,
+ "REFRESH FLOW test_flow",
+ fail("REFRESH FLOW should not evaluate the Spark SQL parsed plan"))
+
assert(refreshWithoutParsedPlan.contains(RefreshTriggeredFlow(Seq("test_flow"))))
+ 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"))
+ Seq("my-flow", "a b").foreach { invalidName =>
+ val invalidRefreshNameError = intercept[IllegalArgumentException] {
+ parseTriggeredFlow(s"REFRESH FLOW $invalidName")
+ }
+ assert(invalidRefreshNameError.getMessage.contains(s"'$invalidName' is
invalid"))
+ }
+
+ 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)
+ Seq("shared_sink", "default.shared_sink",
"spark_catalog.default.shared_sink")
+ .zipWithIndex.foreach { case (targetTable, index) =>
+ val sharedTargetError = intercept[IllegalArgumentException] {
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ s"CREATE FLOW second_flow_$index AS INSERT INTO $targetTable 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" -> "`spark_catalog`.`default`.`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")))
+ spark.table("provider_sink")
+ .selectExpr("part", "input_file_name() AS file")
+ .collect()
+ .foreach { row =>
+ val part = row.getString(0)
+ val file = row.getString(1)
+ assert(file.contains(s"/part=$part/"),
+ s"Expected row for partition '$part' to be stored in its partition
directory: $file")
+ }
+ }
+
+ withTable("options_source", "options_sink") {
+ spark.sql("CREATE TABLE options_source (id STRING, value STRING) USING
parquet")
+ spark.sql("INSERT INTO options_source VALUES ('1', 'a'), ('2', 'b')")
+ spark.sql(
+ "CREATE TABLE options_sink (id STRING, value STRING) USING csv
OPTIONS(delimiter = '|')")
+ val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ spark,
+ "CREATE FLOW options_flow AS INSERT INTO options_sink BY NAME " +
+ "SELECT id, value FROM STREAM(options_source)"),
+ flows)
+ runTriggeredFlow(
+ spark,
+ Seq("options_flow"),
+ RegularTestCase("options", "", ""),
+ Nil,
+ flows)
+ checkAnswer(spark.table("options_sink"), Seq(Row("1", "a"), Row("2",
"b")))
+ }
+
+ withTable("qualified_source") {
+ spark.sql("CREATE TABLE qualified_source (id INT) USING parquet")
+ spark.sql("INSERT INTO qualified_source VALUES (1)")
+ val (firstDb, secondDb) = ("triggered_flow_db1", "triggered_flow_db2")
+ withDatabase(firstDb, secondDb) {
+ spark.sql(s"CREATE DATABASE $firstDb")
+ spark.sql(s"CREATE DATABASE $secondDb")
+ spark.sql(s"CREATE TABLE $firstDb.sink (id INT) USING parquet")
+ spark.sql(s"CREATE TABLE $secondDb.sink (id INT) USING parquet")
+ val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
+
+ spark.sql(s"USE $firstDb")
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ spark,
+ "CREATE FLOW first_qualified_flow AS INSERT INTO sink BY NAME " +
+ "SELECT id FROM STREAM(spark_catalog.default.qualified_source)"),
+ flows)
+
+ spark.sql(s"USE $secondDb")
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ spark,
+ "CREATE FLOW second_qualified_flow AS INSERT INTO sink BY NAME " +
+ "SELECT id FROM STREAM(spark_catalog.default.qualified_source)"),
+ flows)
+ runTriggeredFlow(
+ spark,
+ Seq("first_qualified_flow"),
+ RegularTestCase("qualified-target", "", ""),
+ Nil,
+ flows)
+ runTriggeredFlow(
+ spark,
+ Seq("second_qualified_flow"),
+ RegularTestCase("qualified-target", "", ""),
+ Nil,
+ flows)
+ checkAnswer(spark.table(s"$firstDb.sink"), Row(1))
+ checkAnswer(spark.table(s"$secondDb.sink"), Row(1))
+ }
+ }
+
+ withTable("retry_source", "retry_sink") {
+ spark.sql("CREATE TABLE retry_sink (id INT) USING parquet")
+ val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ spark,
+ "CREATE FLOW retry_flow AS INSERT INTO retry_sink BY NAME " +
+ "SELECT id FROM STREAM(retry_source)"),
+ flows)
+ intercept[AnalysisException] {
+ runTriggeredFlow(
+ spark,
+ Seq("retry_flow"),
+ RegularTestCase("retry", "", ""),
+ Nil,
+ flows)
+ }
+ spark.sql("CREATE TABLE retry_source (id INT) USING parquet")
+ spark.sql("INSERT INTO retry_source VALUES (1)")
+ runTriggeredFlow(
+ spark,
+ Seq("retry_flow"),
+ RegularTestCase("retry", "", ""),
+ Nil,
+ flows)
+ checkAnswer(spark.table("retry_sink"), Row(1))
+ }
+
+ val nonAnsiSession = spark.newSession()
+ nonAnsiSession.conf.set(SQLConf.ANSI_ENABLED.key, false)
+ withTable("non_ansi_source", "non_ansi_sink") {
+ nonAnsiSession.sql("CREATE TABLE non_ansi_source (value STRING) USING
parquet")
+ nonAnsiSession.sql("INSERT INTO non_ansi_source VALUES ('not_an_int')")
+ nonAnsiSession.sql("CREATE TABLE non_ansi_sink (value INT) USING
parquet")
+ val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
+ registerTriggeredFlow(
+ nonAnsiSession,
+ createFlowCommand(
+ nonAnsiSession,
+ "CREATE FLOW non_ansi_flow AS INSERT INTO non_ansi_sink BY NAME " +
+ "SELECT CAST(value AS INT) AS value FROM STREAM(non_ansi_source)"),
+ flows)
+ runTriggeredFlow(
+ nonAnsiSession,
+ Seq("non_ansi_flow"),
+ RegularTestCase("non-ansi", "", ""),
+ Seq(SQLConf.ANSI_ENABLED.key -> "false"),
+ flows)
+ checkAnswer(nonAnsiSession.table("non_ansi_sink"), Row(null))
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
