You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/02/27 09:20:27 UTC
(spark) branch branch-4.0 updated: [SPARK-51322][SQL] Better error message for streaming subquery expression
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new ad174c47f26b [SPARK-51322][SQL] Better error message for streaming subquery expression
ad174c47f26b is described below
commit ad174c47f26b65ef68309781275159440e71a892
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Thu Feb 27 17:19:59 2025 +0800
[SPARK-51322][SQL] Better error message for streaming subquery expression
### What changes were proposed in this pull request?
Today, if a user creates a subquery expression with a streaming query (using the new DataFrame subquery API, or manipulating logical plans directly), he/she will hit weird errors:
- for uncorrelated subquery expressions, Spark will invoke the batch planner to plan the subquery and hit an error because streaming plans are not recognized by the batch planner.
- for correlated subquery expressions in a batch query, it will be rewritten to joins, but the outer batch query is passed to the batch planner and we hit the same error as the previous one
- for correlated subquery expressions in a streaming query, the streaming execution does not go into subqueries to replace `StreamingRelationV2` with `StreamingDataSourceV2ScanRelation`, and after subquery rewriting, the `StreamingRelationV2` will remain and make Spark fail at runtime by `StreamingRelationExec cannot be executed`.
This PR proposes to check streaming subquery expression and fail earlier.
### Why are the changes needed?
better error message
### Does this PR introduce _any_ user-facing change?
Yes, but it's only error message change
### How was this patch tested?
new test
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #50088 from cloud-fan/stream.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit e61a885e217ed5b97e97dee2f311bc51bc85086b)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../src/main/resources/error/error-conditions.json | 5 +++++
.../sql/catalyst/analysis/CheckAnalysis.scala | 3 +++
.../datasources/v2/DataSourceV2Relation.scala | 3 ++-
.../apache/spark/sql/DataFrameSubquerySuite.scala | 24 ++++++++++++++++++++++
4 files changed, 34 insertions(+), 1 deletion(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index a9accd9a97bd..8d149deec98c 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3486,6 +3486,11 @@
"message" : [
"Scalar subquery must return only one column, but got <number>."
]
+ },
+ "STREAMING_QUERY" : {
+ "message" : [
+ "Streaming query is not allowed in subquery expressions."
+ ]
}
},
"sqlState" : "42823"
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index 1b45fcde9126..696c1e10d060 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -1034,6 +1034,9 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
}
def checkSubqueryExpression(plan: LogicalPlan, expr: SubqueryExpression): Unit = {
+ if (expr.plan.isStreaming) {
+ plan.failAnalysis("INVALID_SUBQUERY_EXPRESSION.STREAMING_QUERY", Map.empty)
+ }
checkAnalysis0(expr.plan)
ValidateSubqueryExpression(plan, expr)
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 556283243f63..babdd70d58ba 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -204,7 +204,8 @@ case class StreamingDataSourceV2ScanRelation(
override def name: String = relation.table.name()
override def simpleString(maxFields: Int): String = {
- s"StreamingDataSourceV2ScanRelation${truncatedString(output, "[", ", ", "]", maxFields)} $name"
+ statePrefix + "StreamingDataSourceV2ScanRelation" +
+ s"${truncatedString(output, "[", ", ", "]", maxFields)} $name"
}
override def isStreaming: Boolean = true
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
index 621d468454d4..f5782124243c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
@@ -840,4 +840,28 @@ class DataFrameSubquerySuite extends QueryTest with SharedSparkSession {
checkAnswer(t1.repartition(spark.range(1).select(lit(1)).scalar()), t1)
}
}
+
+ test("SPARK-51322: streaming subquery expression is not allowed") {
+ val rateSource = spark.readStream
+ .format("rate")
+ .option("rowsPerSecond", "10")
+ .option("useManualClock", "true")
+ .load()
+ val df1 = rateSource.select($"value").limit(1)
+ checkError(
+ intercept[AnalysisException](spark.range(5).select(df1.scalar()).collect()),
+ condition = "INVALID_SUBQUERY_EXPRESSION.STREAMING_QUERY"
+ )
+ checkError(
+ intercept[AnalysisException](spark.range(5).where(rateSource.exists()).collect()),
+ condition = "INVALID_SUBQUERY_EXPRESSION.STREAMING_QUERY"
+ )
+
+ // Correlated subquery expression
+ val df2 = rateSource.where($"value" === $"id".outer())
+ checkError(
+ intercept[AnalysisException](spark.range(5).where(df2.exists()).collect()),
+ condition = "INVALID_SUBQUERY_EXPRESSION.STREAMING_QUERY"
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org