This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 605b3ce7dfa5 Revert "[SPARK-60416][SQL][SS] Allow Structured Streaming
Coverage in SQL Tests"
605b3ce7dfa5 is described below
commit 605b3ce7dfa5cc1dcf70d80a9fde92d34e8170f0
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 28 16:54:57 2026 -0700
Revert "[SPARK-60416][SQL][SS] Allow Structured Streaming Coverage in SQL
Tests"
This reverts commit de9d221da902d7391fdb05ab349dd59ebdd46348.
---
.../streaming/available-now-multiple-flows.sql.out | 191 ------
.../available-now-stream-static-join.sql.out | 113 ----
.../streaming/available-now-to-table.sql.out | 99 ---
.../streaming/available-now-union.sql.out | 118 ----
.../available-now-window-aggregation.sql.out | 120 ----
.../streaming/available-now-multiple-flows.sql | 48 --
.../streaming/available-now-stream-static-join.sql | 31 -
.../inputs/streaming/available-now-to-table.sql | 27 -
.../inputs/streaming/available-now-union.sql | 31 -
.../streaming/available-now-window-aggregation.sql | 46 --
.../streaming/available-now-multiple-flows.sql.out | 214 ------
.../available-now-stream-static-join.sql.out | 123 ----
.../streaming/available-now-to-table.sql.out | 111 ---
.../results/streaming/available-now-union.sql.out | 124 ----
.../available-now-window-aggregation.sql.out | 128 ----
.../org/apache/spark/sql/SQLQueryTestHelper.scala | 8 -
.../org/apache/spark/sql/SQLQueryTestSuite.scala | 64 +-
.../spark/sql/StreamingSQLQueryTestSuite.scala | 745 ---------------------
18 files changed, 13 insertions(+), 2328 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
deleted file mode 100644
index 67269e8a29f6..000000000000
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-multiple-flows.sql.out
+++ /dev/null
@@ -1,191 +0,0 @@
--- 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
deleted file mode 100644
index 22e29024268d..000000000000
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-stream-static-join.sql.out
+++ /dev/null
@@ -1,113 +0,0 @@
--- 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
deleted file mode 100644
index b61e35e80a25..000000000000
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-to-table.sql.out
+++ /dev/null
@@ -1,99 +0,0 @@
--- 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
deleted file mode 100644
index 5708eb8b854c..000000000000
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-union.sql.out
+++ /dev/null
@@ -1,118 +0,0 @@
--- 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
deleted file mode 100644
index 4b7977f975b4..000000000000
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/streaming/available-now-window-aggregation.sql.out
+++ /dev/null
@@ -1,120 +0,0 @@
--- 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
deleted file mode 100644
index b89ea6cee5a2..000000000000
---
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-multiple-flows.sql
+++ /dev/null
@@ -1,48 +0,0 @@
-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
deleted file mode 100644
index f16956c2500c..000000000000
---
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-stream-static-join.sql
+++ /dev/null
@@ -1,31 +0,0 @@
-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
deleted file mode 100644
index e70d53f6d283..000000000000
---
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-to-table.sql
+++ /dev/null
@@ -1,27 +0,0 @@
-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
deleted file mode 100644
index c016210499ce..000000000000
---
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-union.sql
+++ /dev/null
@@ -1,31 +0,0 @@
-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
deleted file mode 100644
index 9937e4c83303..000000000000
---
a/sql/core/src/test/resources/sql-tests/inputs/streaming/available-now-window-aggregation.sql
+++ /dev/null
@@ -1,46 +0,0 @@
-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
deleted file mode 100644
index e13458c28b81..000000000000
---
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-multiple-flows.sql.out
+++ /dev/null
@@ -1,214 +0,0 @@
--- 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
deleted file mode 100644
index 5e6218b0abc1..000000000000
---
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-stream-static-join.sql.out
+++ /dev/null
@@ -1,123 +0,0 @@
--- 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
deleted file mode 100644
index 1775d3758e56..000000000000
---
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-to-table.sql.out
+++ /dev/null
@@ -1,111 +0,0 @@
--- 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
deleted file mode 100644
index 9c2e081d4af4..000000000000
---
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-union.sql.out
+++ /dev/null
@@ -1,124 +0,0 @@
--- 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
deleted file mode 100644
index 8ceb34557bb1..000000000000
---
a/sql/core/src/test/resources/sql-tests/results/streaming/available-now-window-aggregation.sql.out
+++ /dev/null
@@ -1,128 +0,0 @@
--- 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 31e7281d31f5..e4782b7fd41e 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,10 +51,6 @@ 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}/")
@@ -97,10 +93,6 @@ 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 03188118bcce..a0b481e1a56c 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,12 +64,6 @@ 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
@@ -193,8 +187,7 @@ 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.
- "streaming/" // Handled by StreamingSQLQueryTestSuite.
+ "ignored.sql" // Do NOT remove this one. It is here to test the ignore
functionality.
) ++ otherIgnoreList
/** List of test cases that require TPCDS table schemas to be loaded. */
private def requireTPCDSCases: Seq[String] = Seq("pipe-operators.sql")
@@ -373,27 +366,6 @@ 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
@@ -410,8 +382,6 @@ 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) {
@@ -421,13 +391,15 @@ class SQLQueryTestSuite extends SharedSparkSession
}
queriesWithDebug.zipWithIndex.foreach { case ((sql, isDebug), i) =>
- lazy val parsedPlan = parsePlan(localSparkSession, sql)
- val customQueryExecuted = executeCustomQuery(sql, () =>
parsedPlan).nonEmpty
- lazy val isCommand = parsedPlan.exists(_.isInstanceOf[Command])
+ val isCommand = try {
+
localSparkSession.sessionState.sqlParser.parsePlan(sql).isInstanceOf[Command]
+ } catch {
+ case _: ParseException => false
+ }
- if (!customQueryExecuted && isCommand) {
+ if (isCommand) {
localSparkSession.sql(sql).collect()
- } else if (!customQueryExecuted && isDebug) {
+ } else if (isDebug) {
// Capture exception stacktrace if the query fails.
var exceptionTrace: Option[String] = None
val (schema, result) = handleExceptions {
@@ -511,29 +483,21 @@ 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 {
- executeCustomQuery(sql, () => parsedPlan)
- .getOrElse(getNormalizedQueryAnalysisResult(localSparkSession,
sql))
- }
+ val (_, output) =
+
handleExceptions(getNormalizedQueryAnalysisResult(localSparkSession, sql))
// We do some query canonicalization now.
AnalyzerOutput(
sql = sql,
schema = None,
output = normalizeTestResults(output.mkString("\n")))
case _ =>
- val (schema, output) = handleExceptions {
- executeCustomQuery(sql, () => parsedPlan)
- .getOrElse(getNormalizedQueryExecutionResult(localSparkSession,
sql))
- }
+ val (schema, output) =
+
handleExceptions(getNormalizedQueryExecutionResult(localSparkSession, sql))
// We do some query canonicalization now.
val executionOutput = ExecutionOutput(
sql = sql,
@@ -612,9 +576,7 @@ class SQLQueryTestSuite extends SharedSparkSession
file.getAbsolutePath.replace(inputFilePath, goldenFilePath) + ".out"
}
- protected lazy val listTestCases: Seq[TestCase] = createTestCases()
-
- protected def createTestCases(): Seq[TestCase] = {
+ protected lazy val listTestCases: 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
deleted file mode 100644
index d0197e330be3..000000000000
---
a/sql/core/src/test/scala/org/apache/spark/sql/StreamingSQLQueryTestSuite.scala
+++ /dev/null
@@ -1,745 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql
-
-import java.nio.charset.StandardCharsets
-import java.util.UUID
-import java.util.concurrent.TimeUnit
-
-import scala.collection.mutable
-
-import org.apache.hadoop.fs.Path
-
-import org.apache.spark.sql.catalyst.analysis.{TableOutputResolver,
UnresolvedIdentifier,
- UnresolvedInsertTarget}
-import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand,
LogicalPlan, UnresolvedInsert}
-import org.apache.spark.sql.catalyst.util.QuotingUtils.quoteIdentifier
-import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
-import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryException,
Trigger}
-import org.apache.spark.tags.ExtendedSQLTest
-import org.apache.spark.util.Utils
-
-// scalastyle:off line.size.limit
-/**
- * SQL golden tests for Structured Streaming flows.
- *
- * CREATE FLOW registers a named flow definition without running it. REFRESH
FLOW runs the
- * remembered definition with Trigger.AvailableNow. The flow must use INSERT
INTO <table> BY NAME,
- * and the target table must already exist. Runs of the same named flow reuse
its checkpoint. The
- * target table must be written only by the flow because batch writes are not
tracked by the
- * streaming sink's metadata log, and each target table may be written by only
one flow. Dropping
- * and recreating the target table while reusing a flow is unsupported and
should not be tested.
- * REFRESH FLOW is test-harness syntax; it is not part of the Spark SQL parser.
- *
- * To run the suite:
- * {{{
- * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite"
- * }}}
- *
- * To run one input file:
- * {{{
- * build/sbt "sql/testOnly org.apache.spark.sql.StreamingSQLQueryTestSuite
-- -z available-now-to-table.sql"
- * }}}
- *
- * To re-generate all execution and analyzer golden files for streaming
inputs, run:
- * {{{
- * SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly
org.apache.spark.sql.StreamingSQLQueryTestSuite"
- * }}}
- *
- * To re-generate both golden files for one streaming input, run:
- * {{{
- * SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly
org.apache.spark.sql.StreamingSQLQueryTestSuite -- -z
available-now-to-table.sql"
- * }}}
- *
- * For example:
- * {{{
- * CREATE TABLE source (id INT, value STRING) USING parquet;
- * INSERT INTO source VALUES (1, 'a'), (2, 'b');
- * CREATE TABLE sink (id INT, value STRING) USING parquet;
- * CREATE FLOW copy_source AS
- * INSERT INTO sink BY NAME
- * SELECT * FROM STREAM(source);
- * REFRESH FLOW copy_source;
- * SELECT * FROM sink ORDER BY id;
- * }}}
- */
-// scalastyle:on line.size.limit
-@ExtendedSQLTest
-class StreamingSQLQueryTestSuite extends SQLQueryTestSuite {
-
- override protected def ignoreList: Set[String] = super.ignoreList -
"streaming/"
-
- override protected lazy val listTestCases: Seq[TestCase] =
- createTestCases().filter(_.name.startsWith("streaming/"))
-
- private def triggeredFlowQueryName(
- testCase: TestCase,
- sparkConfigSet: Seq[(String, String)],
- flowName: Seq[String]): String = {
- val identity =
- Seq(testCase.name, sparkConfigSet.mkString(","),
formatFlowName(flowName)).mkString("\u0000")
- "sql-query-test-" +
UUID.nameUUIDFromBytes(identity.getBytes(StandardCharsets.UTF_8))
- }
-
- private val triggeredFlowTimeoutMs = TimeUnit.MINUTES.toMillis(10)
-
- private case class ResolvedTableName(
- catalog: String,
- namespace: Seq[String],
- table: String) {
- def sql: String = (catalog +: namespace :+
table).map(quoteIdentifier).mkString(".")
- }
-
- private case class TriggeredFlow(
- flowName: Seq[String],
- query: LogicalPlan,
- targetTable: String,
- resolvedTargetTable: ResolvedTableName,
- checkpointLocation: String)
-
- private sealed trait TriggeredFlowStatement
- private case class CreateTriggeredFlow(command: CreateFlowCommand)
- extends TriggeredFlowStatement
- private case class RefreshTriggeredFlow(flowName: Seq[String])
- extends TriggeredFlowStatement
-
- private def formatFlowName(flowName: Seq[String]): String =
- flowName.map(quoteIdentifier).mkString(".")
-
- private def sameFlowName(
- session: SparkSession,
- left: Seq[String],
- right: Seq[String]): Boolean = {
- left.length == right.length && left.zip(right).forall { case (l, r) =>
- session.sessionState.conf.resolver(l, r)
- }
- }
-
- private def loadV1Table(session: SparkSession, tableName: String):
Option[CatalogTable] = {
- val analyzer = session.sessionState.analyzer
- import analyzer.CatalogAndIdentifier
-
- val multipartIdentifier =
session.sessionState.sqlParser.parseMultipartIdentifier(tableName)
- val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier
- catalog.asTableCatalog.loadTable(identifier) match {
- case table: V1Table => Some(table.v1Table)
- case table: V2TableWithV1Fallback => Some(table.v1Table)
- case _ => None
- }
- }
-
- private def resolveTableName(
- session: SparkSession,
- multipartIdentifier: Seq[String]): ResolvedTableName = {
- val analyzer = session.sessionState.analyzer
- import analyzer.CatalogAndIdentifier
-
- val CatalogAndIdentifier(catalog, identifier) = multipartIdentifier
- ResolvedTableName(catalog.name(), identifier.namespace().toSeq,
identifier.name())
- }
-
- private def sameTableName(
- session: SparkSession,
- left: ResolvedTableName,
- right: ResolvedTableName): Boolean = {
- val resolver = session.sessionState.conf.resolver
- resolver(left.catalog, right.catalog) &&
- left.namespace.length == right.namespace.length &&
- left.namespace.zip(right.namespace).forall { case (l, r) => resolver(l,
r) } &&
- resolver(left.table, right.table)
- }
-
- private def parseCreateTriggeredFlow(
- session: SparkSession,
- command: CreateFlowCommand): TriggeredFlow = {
- val flowName = command.name match {
- case identifier: UnresolvedIdentifier => identifier.nameParts
- case _ =>
- throw new IllegalArgumentException(
- "SQL golden triggered flows require a static flow name.")
- }
- val insert = command.flowOperation match {
- case insert: UnresolvedInsert => insert
- case _ =>
- throw new IllegalArgumentException(
- "SQL golden triggered flows require CREATE FLOW AS " +
- "INSERT INTO ... BY NAME SELECT.")
- }
- require(insert.byName && !insert.overwrite && insert.partitionSpec.isEmpty
&&
- insert.userSpecifiedCols.isEmpty && insert.replaceCriteriaOpt.isEmpty
&&
- !insert.withSchemaEvolution,
- "SQL golden triggered flows require CREATE FLOW AS INSERT INTO ... BY
NAME SELECT.")
-
- val (targetTable, resolvedTargetTable) = insert.table match {
- case target: UnresolvedInsertTarget =>
- val targetTable =
target.multipartIdentifier.map(quoteIdentifier).mkString(".")
- targetTable -> resolveTableName(session, target.multipartIdentifier)
- case _ =>
- throw new IllegalArgumentException(
- "SQL golden triggered flows require a static table identifier as the
INSERT target.")
- }
- val checkpointLocation = new Path(
- session.sessionState.conf.warehousePath,
- s"sql-query-test-checkpoint-${UUID.randomUUID()}").toString
- TriggeredFlow(
- flowName,
- insert.query,
- targetTable,
- resolvedTargetTable,
- checkpointLocation)
- }
-
- private def registerTriggeredFlow(
- session: SparkSession,
- command: CreateFlowCommand,
- flows: mutable.Map[Seq[String], TriggeredFlow]): TriggeredFlow = {
- session.withActive {
- val flow = parseCreateTriggeredFlow(session, command)
- require(!flows.keys.exists(sameFlowName(session, _, flow.flowName)),
- s"Triggered flow ${formatFlowName(flow.flowName)} is already defined
in this test run.")
- flows.values.find { registeredFlow =>
- sameTableName(session, registeredFlow.resolvedTargetTable,
flow.resolvedTargetTable)
- }.foreach { registeredFlow =>
- throw new IllegalArgumentException(
- s"SQL golden triggered flow target table '${flow.targetTable}' is
already written by " +
- s"flow ${formatFlowName(registeredFlow.flowName)}.")
- }
- flows += flow.flowName -> flow
- flow
- }
- }
-
- private def resolveTriggeredFlow(
- session: SparkSession,
- flowName: Seq[String],
- flows: mutable.Map[Seq[String], TriggeredFlow]): TriggeredFlow = {
- flows.collectFirst {
- case (registeredName, flow) if sameFlowName(session, registeredName,
flowName) => flow
- }.getOrElse {
- throw new IllegalArgumentException(
- s"Triggered flow ${formatFlowName(flowName)} has not been defined in
this test run.")
- }
- }
-
- private def parseRefreshTriggeredFlowStatement(
- session: SparkSession,
- name: String): RefreshTriggeredFlow = {
- try {
-
RefreshTriggeredFlow(session.sessionState.sqlParser.parseMultipartIdentifier(name))
- } catch {
- case e: ParseException =>
- throw new IllegalArgumentException(
- s"REFRESH FLOW requires a valid flow name, but '$name' is invalid.",
e)
- }
- }
-
- private def parseTriggeredFlowStatement(
- session: SparkSession,
- sql: String,
- parsedPlan: => Option[LogicalPlan]): Option[TriggeredFlowStatement] = {
- session.withActive {
- val tokens = sql.trim.split("\\s+", 3)
- if (tokens.length >= 2 && tokens(1).equalsIgnoreCase("FLOW")) {
- if (tokens(0).equalsIgnoreCase("REFRESH")) {
- if (tokens.length == 2) {
- throw new IllegalArgumentException("REFRESH FLOW requires a flow
name.")
- }
- Some(parseRefreshTriggeredFlowStatement(session, tokens(2)))
- } else if (tokens(0).equalsIgnoreCase("CREATE")) {
- parsedPlan.collect {
- case command: CreateFlowCommand => CreateTriggeredFlow(command)
- }
- } else {
- None
- }
- } else {
- None
- }
- }
- }
-
- private def parseTriggeredFlowStatement(
- session: SparkSession,
- sql: String): Option[TriggeredFlowStatement] = {
- parseTriggeredFlowStatement(session, sql, parsePlan(session, sql))
- }
-
- /** Runs one REFRESH FLOW statement to completion. */
- private def runTriggeredFlow(
- session: SparkSession,
- flowName: Seq[String],
- testCase: TestCase,
- sparkConfigSet: Seq[(String, String)],
- flows: mutable.Map[Seq[String], TriggeredFlow]): (String, Seq[String]) =
{
- session.withActive {
- val flow = resolveTriggeredFlow(session, flowName, flows)
- val targetTableName = flow.resolvedTargetTable.sql
- val df = classic.Dataset.ofRows(
- session.asInstanceOf[classic.SparkSession], flow.query.clone())
- require(df.isStreaming,
- s"SQL golden triggered flow ${formatFlowName(flow.flowName)} requires
a streaming query, " +
- "but its CREATE FLOW query reads no streaming source.")
- require(session.catalog.tableExists(targetTableName),
- s"SQL golden triggered flow target table '${flow.targetTable}' does
not exist.")
- val queryName = triggeredFlowQueryName(testCase, sparkConfigSet,
flow.flowName)
- val targetTable = loadV1Table(session, targetTableName)
- val alignedQuery = TableOutputResolver.resolveOutputColumns(
- targetTableName,
- session.table(targetTableName).queryExecution.analyzed.output,
- df.queryExecution.analyzed,
- byName = true,
- session.sessionState.conf,
- TableOutputResolver.DefaultValueFillMode.FILL)
- val alignedDf = classic.Dataset.ofRows(
- session.asInstanceOf[classic.SparkSession], alignedQuery)
-
- val writer = alignedDf.writeStream
- .outputMode(OutputMode.Append())
- .trigger(Trigger.AvailableNow())
- .queryName(queryName)
- .option("checkpointLocation", flow.checkpointLocation)
- targetTable.foreach { table =>
- writer.format(table.provider.getOrElse {
- throw new IllegalArgumentException(
- s"SQL golden triggered flow target table '${flow.targetTable}' has
no provider.")
- })
- writer.options(table.storage.properties)
- if (table.partitionColumnNames.nonEmpty) {
- writer.partitionBy(table.partitionColumnNames: _*)
- }
- }
- val query = writer.toTable(targetTableName)
- Utils.tryWithSafeFinally {
- try {
- if (!query.awaitTermination(triggeredFlowTimeoutMs)) {
- throw new IllegalStateException(
- s"SQL golden triggered flow ${formatFlowName(flow.flowName)} did
not terminate " +
- s"within
${TimeUnit.MILLISECONDS.toMinutes(triggeredFlowTimeoutMs)} minutes.")
- }
- } catch {
- case e: StreamingQueryException if e.cause != null => throw e.cause
- }
- } {
- query.stop()
- }
-
- (emptySchema, Seq.empty)
- }
- }
-
- override protected def createCustomQueryExecutor(
- session: SparkSession,
- testCase: TestCase,
- sparkConfigSet: Seq[(String, String)],
- analyzerOnly: Boolean):
- (String, () => Option[LogicalPlan]) => Option[(String, Seq[String])] = {
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- (sql, getParsedPlan) => parseTriggeredFlowStatement(session, sql,
getParsedPlan()).map {
- case CreateTriggeredFlow(command) =>
- val flow = registerTriggeredFlow(session, command, flows)
- if (analyzerOnly) {
- val df = classic.Dataset.ofRows(
- session.asInstanceOf[classic.SparkSession], flow.query)
- getNormalizedQueryAnalysisResult(df)
- } else {
- (emptySchema, Seq.empty)
- }
- case RefreshTriggeredFlow(flowName) =>
- if (analyzerOnly) {
- resolveTriggeredFlow(session, flowName, flows)
- (emptySchema, Seq.empty)
- } else {
- runTriggeredFlow(session, flowName, testCase, sparkConfigSet, flows)
- }
- }
- }
-
- test("test triggered flow parsing") {
- def querySpecs(input: String): Seq[(String, Boolean)] = {
- val (comments, code) = splitCommentsAndCodes(input)
- getQueriesWithDebugFlag(code, comments, Nil)
- }
-
- def parseTriggeredFlow(sql: String): Option[TriggeredFlowStatement] = {
- parseTriggeredFlowStatement(spark, sql)
- }
-
- def createFlowCommand(sql: String): CreateFlowCommand = {
- parseTriggeredFlow(sql) match {
- case Some(CreateTriggeredFlow(command)) => command
- case statement => fail(s"Expected CREATE FLOW, found $statement")
- }
- }
-
- def refreshFlowName(sql: String): Seq[String] = {
- parseTriggeredFlow(sql) match {
- case Some(RefreshTriggeredFlow(flowName)) => flowName
- case statement => fail(s"Expected REFRESH FLOW, found $statement")
- }
- }
-
- val specs = querySpecs(
- """-- ordinary comment
- |SELECT 1;
- |CREATE FLOW test_flow AS
- |INSERT INTO `spark_catalog`.`default`.`sink-table` BY NAME
- |SELECT * FROM STREAM(source);
- |REFRESH FLOW test_flow;
- |SELECT 2;
- |""".stripMargin)
- assert(specs.map(_._1) == Seq(
- "SELECT 1",
- """CREATE FLOW test_flow AS
- |INSERT INTO `spark_catalog`.`default`.`sink-table` BY NAME
- |SELECT * FROM STREAM(source)""".stripMargin,
- "REFRESH FLOW test_flow",
- "SELECT 2"))
-
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(spark, createFlowCommand(specs(1)._1), flows)
- val flowName = refreshFlowName(specs(2)._1)
- val flow = resolveTriggeredFlow(spark, flowName, flows)
- assert(flow.targetTable == "`spark_catalog`.`default`.`sink-table`")
- assert(resolveTriggeredFlow(spark, flowName, flows) eq flow)
- val separatelyRegisteredFlow = registerTriggeredFlow(
- spark,
- createFlowCommand(specs(1)._1),
- mutable.Map.empty)
- assert(separatelyRegisteredFlow.checkpointLocation !=
flow.checkpointLocation)
- assert(parseTriggeredFlow(specs.head._1).isEmpty)
- assert(parseTriggeredFlow("select interval 10 nanoseconds").isEmpty)
- assert(parseTriggeredFlow(specs(1)._1)
- .exists(_.isInstanceOf[CreateTriggeredFlow]))
- assert(parseTriggeredFlow(specs(2)._1)
- .exists(_.isInstanceOf[RefreshTriggeredFlow]))
- val refreshWithoutParsedPlan = parseTriggeredFlowStatement(
- spark,
- "REFRESH FLOW test_flow",
- fail("REFRESH FLOW should not evaluate the Spark SQL parsed plan"))
-
assert(refreshWithoutParsedPlan.contains(RefreshTriggeredFlow(Seq("test_flow"))))
- assert(refreshFlowName("refresh flow `catalog`.`flow-name`") ==
Seq("catalog", "flow-name"))
- val missingRefreshNameError = intercept[IllegalArgumentException] {
- parseTriggeredFlow("REFRESH FLOW")
- }
- assert(missingRefreshNameError.getMessage.contains("requires a flow name"))
- Seq("my-flow", "a b").foreach { invalidName =>
- val invalidRefreshNameError = intercept[IllegalArgumentException] {
- parseTriggeredFlow(s"REFRESH FLOW $invalidName")
- }
- assert(invalidRefreshNameError.getMessage.contains(s"'$invalidName' is
invalid"))
- }
-
- val debugSpec = querySpecs(
- """--DEBUG
- |CREATE FLOW debug_flow AS
- |INSERT INTO sink BY NAME SELECT * FROM STREAM(source);
- |""".stripMargin).head
- assert(debugSpec._2)
-
- val nonByNameError = intercept[IllegalArgumentException] {
- registerTriggeredFlow(
- spark,
- createFlowCommand("CREATE FLOW f AS INSERT INTO sink SELECT * FROM
STREAM(source)"),
- mutable.Map.empty)
- }
- assert(nonByNameError.getMessage.contains("INSERT INTO ... BY NAME
SELECT"))
-
- val undefinedFlowError = intercept[IllegalArgumentException] {
- resolveTriggeredFlow(spark, Seq("missing_flow"), mutable.Map.empty)
- }
- assert(undefinedFlowError.getMessage.contains("has not been defined in
this test run"))
-
- val batchFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- "CREATE FLOW batch_flow AS INSERT INTO sink BY NAME SELECT 1 AS id"),
- batchFlows)
- val batchFlowError = intercept[IllegalArgumentException] {
- runTriggeredFlow(
- spark,
- Seq("batch_flow"),
- RegularTestCase("batch-flow-error", "", ""),
- Nil,
- batchFlows)
- }
- assert(batchFlowError.getMessage.contains("triggered flow `batch_flow`"))
- assert(!batchFlowError.getMessage.contains("REFRESH FLOW"))
-
- val sharedTargetFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- "CREATE FLOW first_flow AS INSERT INTO shared_sink BY NAME " +
- "SELECT * FROM STREAM(first_source)"),
- sharedTargetFlows)
- Seq("shared_sink", "default.shared_sink",
"spark_catalog.default.shared_sink")
- .zipWithIndex.foreach { case (targetTable, index) =>
- val sharedTargetError = intercept[IllegalArgumentException] {
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- s"CREATE FLOW second_flow_$index AS INSERT INTO $targetTable BY
NAME " +
- "SELECT * FROM STREAM(second_source)"),
- sharedTargetFlows)
- }
- assert(sharedTargetError.getMessage.contains("already written by flow
`first_flow`"))
- }
-
- withTable("by_name_source", "by_name_sink") {
- spark.sql("CREATE TABLE by_name_source (second INT, first INT) USING
parquet")
- spark.sql("INSERT INTO by_name_source VALUES (2, 1)")
- spark.sql("CREATE TABLE by_name_sink (first INT, second INT) USING
parquet")
- val byNameFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- """CREATE FLOW by_name_flow AS
- |INSERT INTO by_name_sink BY NAME
- |SELECT second, first FROM STREAM(by_name_source)
- |""".stripMargin),
- byNameFlows)
- runTriggeredFlow(
- spark,
- Seq("by_name_flow"),
- RegularTestCase("by-name", "", ""),
- Nil,
- byNameFlows)
- checkAnswer(spark.table("by_name_sink"), Row(1, 2))
- }
-
- withTable("incompatible_source", "incompatible_sink") {
- spark.sql("CREATE TABLE incompatible_source (id INT) USING parquet")
- spark.sql("CREATE TABLE incompatible_sink (other INT) USING parquet")
- val incompatibleFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- """CREATE FLOW incompatible_flow AS
- |INSERT INTO incompatible_sink BY NAME
- |SELECT id FROM STREAM(incompatible_source)
- |""".stripMargin),
- incompatibleFlows)
- checkError(
- exception = intercept[AnalysisException] {
- runTriggeredFlow(
- spark,
- Seq("incompatible_flow"),
- RegularTestCase("incompatible", "", ""),
- Nil,
- incompatibleFlows)
- },
- condition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS",
- parameters = Map(
- "tableName" -> "`spark_catalog`.`default`.`incompatible_sink`",
- "extraColumns" -> "`id`"))
- }
-
- withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
- withTable("runtime_error_source", "runtime_error_sink") {
- spark.sql("CREATE TABLE runtime_error_source (value STRING) USING
parquet")
- spark.sql("INSERT INTO runtime_error_source VALUES ('not_an_int')")
- spark.sql("CREATE TABLE runtime_error_sink (value INT) USING parquet")
- val runtimeErrorFlows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- """CREATE FLOW runtime_error_flow AS
- |INSERT INTO runtime_error_sink BY NAME
- |SELECT CAST(value AS INT) AS value FROM
STREAM(runtime_error_source)
- |""".stripMargin),
- runtimeErrorFlows)
- val (_, runtimeErrorOutput) = handleExceptions {
- runTriggeredFlow(
- spark,
- Seq("runtime_error_flow"),
- RegularTestCase("runtime-error", "", ""),
- Nil,
- runtimeErrorFlows)
- }
- assert(runtimeErrorOutput.exists(_.contains("CAST_INVALID_INPUT")))
- assert(!runtimeErrorOutput.exists(_.contains("STREAM_FAILED")))
- assert(!runtimeErrorOutput.exists(_.contains("runId")))
- }
- }
- }
-
- test("test triggered flow execution semantics") {
- def createFlowCommand(session: SparkSession, sql: String):
CreateFlowCommand = {
- parseTriggeredFlowStatement(session, sql) match {
- case Some(CreateTriggeredFlow(command)) => command
- case statement => fail(s"Expected CREATE FLOW, found $statement")
- }
- }
-
- withTable("provider_source", "provider_sink") {
- spark.sql("CREATE TABLE provider_source (id INT, part STRING) USING
parquet")
- spark.sql("INSERT INTO provider_source VALUES (1, 'a'), (2, 'b')")
- spark.sql(
- "CREATE TABLE provider_sink (id INT, part STRING) USING JSON
PARTITIONED BY (part)")
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- spark,
- "CREATE FLOW provider_flow AS INSERT INTO provider_sink BY NAME " +
- "SELECT id, part FROM STREAM(provider_source)"),
- flows)
- runTriggeredFlow(
- spark,
- Seq("provider_flow"),
- RegularTestCase("provider", "", ""),
- Nil,
- flows)
- checkAnswer(spark.table("provider_sink"), Seq(Row(1, "a"), Row(2, "b")))
- spark.table("provider_sink")
- .selectExpr("part", "input_file_name() AS file")
- .collect()
- .foreach { row =>
- val part = row.getString(0)
- val file = row.getString(1)
- assert(file.contains(s"/part=$part/"),
- s"Expected row for partition '$part' to be stored in its partition
directory: $file")
- }
- }
-
- withTable("options_source", "options_sink") {
- spark.sql("CREATE TABLE options_source (id STRING, value STRING) USING
parquet")
- spark.sql("INSERT INTO options_source VALUES ('1', 'a'), ('2', 'b')")
- spark.sql(
- "CREATE TABLE options_sink (id STRING, value STRING) USING csv
OPTIONS(delimiter = '|')")
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- spark,
- "CREATE FLOW options_flow AS INSERT INTO options_sink BY NAME " +
- "SELECT id, value FROM STREAM(options_source)"),
- flows)
- runTriggeredFlow(
- spark,
- Seq("options_flow"),
- RegularTestCase("options", "", ""),
- Nil,
- flows)
- checkAnswer(spark.table("options_sink"), Seq(Row("1", "a"), Row("2",
"b")))
- }
-
- withTable("qualified_source") {
- spark.sql("CREATE TABLE qualified_source (id INT) USING parquet")
- spark.sql("INSERT INTO qualified_source VALUES (1)")
- val (firstDb, secondDb) = ("triggered_flow_db1", "triggered_flow_db2")
- withDatabase(firstDb, secondDb) {
- spark.sql(s"CREATE DATABASE $firstDb")
- spark.sql(s"CREATE DATABASE $secondDb")
- spark.sql(s"CREATE TABLE $firstDb.sink (id INT) USING parquet")
- spark.sql(s"CREATE TABLE $secondDb.sink (id INT) USING parquet")
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
-
- spark.sql(s"USE $firstDb")
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- spark,
- "CREATE FLOW first_qualified_flow AS INSERT INTO sink BY NAME " +
- "SELECT id FROM STREAM(spark_catalog.default.qualified_source)"),
- flows)
-
- spark.sql(s"USE $secondDb")
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- spark,
- "CREATE FLOW second_qualified_flow AS INSERT INTO sink BY NAME " +
- "SELECT id FROM STREAM(spark_catalog.default.qualified_source)"),
- flows)
- runTriggeredFlow(
- spark,
- Seq("first_qualified_flow"),
- RegularTestCase("qualified-target", "", ""),
- Nil,
- flows)
- runTriggeredFlow(
- spark,
- Seq("second_qualified_flow"),
- RegularTestCase("qualified-target", "", ""),
- Nil,
- flows)
- checkAnswer(spark.table(s"$firstDb.sink"), Row(1))
- checkAnswer(spark.table(s"$secondDb.sink"), Row(1))
- }
- }
-
- withTable("retry_source", "retry_sink") {
- spark.sql("CREATE TABLE retry_sink (id INT) USING parquet")
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- spark,
- createFlowCommand(
- spark,
- "CREATE FLOW retry_flow AS INSERT INTO retry_sink BY NAME " +
- "SELECT id FROM STREAM(retry_source)"),
- flows)
- intercept[AnalysisException] {
- runTriggeredFlow(
- spark,
- Seq("retry_flow"),
- RegularTestCase("retry", "", ""),
- Nil,
- flows)
- }
- spark.sql("CREATE TABLE retry_source (id INT) USING parquet")
- spark.sql("INSERT INTO retry_source VALUES (1)")
- runTriggeredFlow(
- spark,
- Seq("retry_flow"),
- RegularTestCase("retry", "", ""),
- Nil,
- flows)
- checkAnswer(spark.table("retry_sink"), Row(1))
- }
-
- val nonAnsiSession = spark.newSession()
- nonAnsiSession.conf.set(SQLConf.ANSI_ENABLED.key, false)
- withTable("non_ansi_source", "non_ansi_sink") {
- nonAnsiSession.sql("CREATE TABLE non_ansi_source (value STRING) USING
parquet")
- nonAnsiSession.sql("INSERT INTO non_ansi_source VALUES ('not_an_int')")
- nonAnsiSession.sql("CREATE TABLE non_ansi_sink (value INT) USING
parquet")
- val flows = mutable.Map.empty[Seq[String], TriggeredFlow]
- registerTriggeredFlow(
- nonAnsiSession,
- createFlowCommand(
- nonAnsiSession,
- "CREATE FLOW non_ansi_flow AS INSERT INTO non_ansi_sink BY NAME " +
- "SELECT CAST(value AS INT) AS value FROM STREAM(non_ansi_source)"),
- flows)
- runTriggeredFlow(
- nonAnsiSession,
- Seq("non_ansi_flow"),
- RegularTestCase("non-ansi", "", ""),
- Seq(SQLConf.ANSI_ENABLED.key -> "false"),
- flows)
- checkAnswer(nonAnsiSession.table("non_ansi_sink"), Row(null))
- }
- }
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
