This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8622bf64d253 [SPARK-60416][SQL][SS] Allow Structured Streaming
Coverage in SQL Tests
8622bf64d253 is described below
commit 8622bf64d253128af42121f0da257c055961a627
Author: Siying Dong <[email protected]>
AuthorDate: Fri Oct 30 16:23:03 2026 +0000
[SPARK-60416][SQL][SS] Allow Structured Streaming Coverage in SQL Tests
### What changes were proposed in this pull request?
This PR relands #59665, which was reverted by commit
`6cd6698511134afe5f90694ae8e290235f6b8108` after post-merge CI failures.
It adds `StreamingSQLQueryTestSuite`, a `SQLQueryTestSuite` subclass that
lets SQL golden tests
exercise Structured Streaming flows. The test harness:
- 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.
This reland also fixes the post-merge failure by accepting both
`InsertIntoStatement` and
`UnresolvedInsert` flow operations, and both `UnresolvedRelation` and
`UnresolvedInsertTarget`
targets. This matches the dual-shape handling used by pipeline SQL
registration after #59750.
### Why are the changes needed?
SQL golden tests cannot normally execute a query containing streaming
sources because 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.
The original PR was tested on a base from before #59750. That parser
follow-up restored
`InsertIntoStatement` as the parsed plan for static INSERT targets. The
original change later
merged cleanly at the text level, but its harness still matched only
`UnresolvedInsert`, causing 12
`StreamingSQLQueryTestSuite` cases to fail in post-merge CI. Supporting
both parser plan shapes
fixes that semantic integration issue.
### 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 original PR passed these focused tests before the parser follow-up was
merged:
```
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.
For this reland, the focused parser test was attempted locally, but SBT
could not resolve the Maven
repository hosts while downloading the newly required build dependencies,
so the build stopped
before project loading. The compatibility path is exercised by `test
triggered flow parsing` and
all ten AvailableNow execution/analyzer golden tests in PR CI.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #59935 from siying/codex/streaming-sql-available-now-workloads.
Authored-by: Siying Dong <[email protected]>
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 | 65 +-
.../spark/sql/StreamingSQLQueryTestSuite.scala | 823 +++++++++++++++++++++
18 files changed, 2407 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..9d193a591372 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,14 @@ 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)
+ // Custom statements are setup commands, so run them regardless of their
--DEBUG marker.
+ 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 +512,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 +613,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..ff674468475c
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala
@@ -0,0 +1,823 @@
+/*
+ * 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, UnresolvedRelation}
+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,
InsertIntoStatement,
+ 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,
+ registrationCatalog: String,
+ registrationNamespace: Seq[String],
+ 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: InsertIntoStatement => insert
+ case insert: UnresolvedInsert =>
insert.toInsertIntoStatement(insert.table)
+ 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 targetIdentifier = insert.table match {
+ case target: UnresolvedInsertTarget => target.multipartIdentifier
+ case target: UnresolvedRelation => target.multipartIdentifier
+ case _ =>
+ throw new IllegalArgumentException(
+ "SQL golden triggered flows require a static table identifier as the
INSERT target.")
+ }
+ val targetTable = targetIdentifier.map(quoteIdentifier).mkString(".")
+ val resolvedTargetTable = resolveTableName(session, targetIdentifier)
+ val checkpointLocation = new Path(
+ session.sessionState.conf.warehousePath,
+ s"sql-query-test-checkpoint-${UUID.randomUUID()}").toString
+ val catalogManager = session.sessionState.catalogManager
+ TriggeredFlow(
+ flowName,
+ insert.query,
+ targetTable,
+ resolvedTargetTable,
+ catalogManager.currentCatalog.name(),
+ catalogManager.currentNamespace.toSeq,
+ checkpointLocation)
+ }
+
+ private def withFlowRegistrationContext[T](
+ session: SparkSession,
+ flow: TriggeredFlow)(f: => T): T = {
+ session.withActive {
+ val catalogManager = session.sessionState.catalogManager
+ val previousCatalog = catalogManager.currentCatalog.name()
+ val previousNamespace = catalogManager.currentNamespace
+ Utils.tryWithSafeFinally {
+ catalogManager.setCurrentCatalog(flow.registrationCatalog)
+ catalogManager.setCurrentNamespace(flow.registrationNamespace.toArray)
+ f
+ } {
+ catalogManager.setCurrentCatalog(previousCatalog)
+ catalogManager.setCurrentNamespace(previousNamespace)
+ }
+ }
+ }
+
+ 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(0).equalsIgnoreCase("REFRESH") &&
+ tokens(1).equalsIgnoreCase("FLOW")) {
+ if (tokens.length == 2) {
+ throw new IllegalArgumentException("REFRESH FLOW requires a flow
name.")
+ }
+ Some(parseRefreshTriggeredFlowStatement(session, tokens(2)))
+ } else {
+ parsedPlan.collect {
+ case command: CreateFlowCommand => CreateTriggeredFlow(command)
+ }
+ }
+ }
+ }
+
+ 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
+ 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 alignedDf = withFlowRegistrationContext(session, flow) {
+ 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.")
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
+ targetTableName,
+ session.table(targetTableName).queryExecution.analyzed.output,
+ df.queryExecution.analyzed,
+ byName = true,
+ session.sessionState.conf,
+ TableOutputResolver.DefaultValueFillMode.FILL)
+ classic.Dataset.ofRows(
+ session.asInstanceOf[classic.SparkSession], alignedQuery)
+ }
+
+ val writer = alignedDf.writeStream
+ .outputMode(OutputMode.Append())
+ .trigger(Trigger.AvailableNow())
+ 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
+ .queryName(queryName)
+ .option("checkpointLocation", flow.checkpointLocation)
+ .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) {
+ withFlowRegistrationContext(session, flow) {
+ 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(specs(1)._1)
+ .exists(_.isInstanceOf[CreateTriggeredFlow]))
+ assert(parseTriggeredFlow(specs(2)._1)
+ .exists(_.isInstanceOf[RefreshTriggeredFlow]))
+ Seq(
+ "CREATE /* comment */ FLOW commented_flow AS " +
+ "INSERT INTO sink BY NAME SELECT * FROM STREAM(source)",
+ "/* comment */ CREATE FLOW leading_comment_flow AS " +
+ "INSERT INTO sink BY NAME SELECT * FROM STREAM(source)"
+ ).foreach { commentedCreateFlow =>
+ assert(parseTriggeredFlow(commentedCreateFlow)
+ .exists(_.isInstanceOf[CreateTriggeredFlow]))
+ }
+ 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 dynamicTargetCommand = createFlowCommand(
+ "CREATE FLOW dynamic_target_flow AS " +
+ "INSERT INTO IDENTIFIER('sink' || '_table') BY NAME " +
+ "SELECT * FROM STREAM(source)")
+ assert(dynamicTargetCommand.flowOperation.isInstanceOf[UnresolvedInsert])
+ val dynamicTargetError = intercept[IllegalArgumentException] {
+ registerTriggeredFlow(spark, dynamicTargetCommand, mutable.Map.empty)
+ }
+ assert(dynamicTargetError.getMessage.contains("static table identifier"))
+
+ val undefinedFlowError = intercept[IllegalArgumentException] {
+ resolveTriggeredFlow(spark, Seq("missing_flow"), mutable.Map.empty)
+ }
+ assert(undefinedFlowError.getMessage.contains("has not been defined in
this test run"))
+
+ withTable("batch_sink") {
+ spark.sql("CREATE TABLE batch_sink (id INT) USING parquet")
+ val batchFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ "CREATE FLOW batch_flow AS INSERT INTO batch_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` requires a streaming query"))
+ 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))
+ }
+ }
+
+ val (sourceDb, refreshDb) = ("triggered_flow_source_db",
"triggered_flow_refresh_db")
+ withDatabase(sourceDb, refreshDb) {
+ spark.sql(s"CREATE DATABASE $sourceDb")
+ spark.sql(s"CREATE DATABASE $refreshDb")
+ spark.sql(s"CREATE TABLE $sourceDb.source (id INT) USING parquet")
+ spark.sql(s"INSERT INTO $sourceDb.source VALUES (1)")
+ spark.sql(s"CREATE TABLE $sourceDb.sink (id INT) USING parquet")
+ spark.sql(s"CREATE TABLE $refreshDb.source (id INT) USING parquet")
+ spark.sql(s"INSERT INTO $refreshDb.source VALUES (2)")
+ val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
+
+ spark.sql(s"USE $sourceDb")
+ registerTriggeredFlow(
+ spark,
+ createFlowCommand(
+ spark,
+ "CREATE FLOW source_context_flow AS INSERT INTO sink BY NAME " +
+ "SELECT id FROM STREAM(source)"),
+ flows)
+ spark.sql(s"USE $refreshDb")
+ runTriggeredFlow(
+ spark,
+ Seq("source_context_flow"),
+ RegularTestCase("source-context", "", ""),
+ Nil,
+ flows)
+ assert(spark.catalog.currentDatabase == refreshDb)
+ checkAnswer(spark.table(s"$sourceDb.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]
