You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "AnishMahto (via GitHub)" <gi...@apache.org> on 2025/06/17 22:37:22 UTC
[PR] [SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
AnishMahto opened a new pull request, #51208:
URL: https://github.com/apache/spark/pull/51208
### What changes were proposed in this pull request?
Validate that streaming flows are actually backed by streaming sources, and batch flows are actually backed by batch sources. Also improve SDP test harnesses to be explicit about whether a streaming table or materialized view is being created, to match the Python/SQL API.
### Why are the changes needed?
This change helps prevent incorrect usage of streaming/batch flows, such as directly reading from a batch source from a streaming table's flow. In this case for example, the `STREAM` key word to mark a SQL batch source as streaming or `readStream` should be used in Python to stream read from an otherwise non-streaming file source.
### Does this PR introduce _any_ user-facing change?
No, as this impacts SDP which is not released in any Spark version yet.
### How was this patch tested?
Existing suites and added tests to `ConnectInvalidPipelineSuite`
### Was this patch authored or co-authored using generative AI tooling?
No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52591][SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #51208:
URL: https://github.com/apache/spark/pull/51208#issuecomment-3017270831
@sryza I indicated that JIRA ticket for this PR does not have the fixed version as well as the PR link. Could you please check this? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on PR #51208:
URL: https://github.com/apache/spark/pull/51208#issuecomment-2982031499
@sryza
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-52591][SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza commented on PR #51208:
URL: https://github.com/apache/spark/pull/51208#issuecomment-3019581486
@HeartSaVioR 👍 just added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
Posted by "sryza (via GitHub)" <gi...@apache.org>.
sryza commented on code in PR #51208:
URL: https://github.com/apache/spark/pull/51208#discussion_r2159723113
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -55,6 +55,75 @@ trait GraphValidations extends Logging {
multiQueryTables
}
+ protected[graph] def validateFlowStreamingness(): Unit = {
+ flowsTo.foreach { case (destTableIdentifier, flows) =>
+ val destTableOpt = table.get(destTableIdentifier)
+
+ // If the destination identifier does not correspond to a table, it must be a view.
+ val destViewOpt = destTableOpt.fold(view.get(destTableIdentifier))(_ => None)
Review Comment:
Why not just use `view.get(destTableIdentifier)` here?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -55,6 +55,75 @@ trait GraphValidations extends Logging {
multiQueryTables
}
+ protected[graph] def validateFlowStreamingness(): Unit = {
Review Comment:
This could use some docstring. It doesn't need to go into the individual cases (because that could get out of sync easily), but could be at the level of "Some dataset types require that the flows that target them are backed by streaming queries. Others require batch queries. This validates that the dataset types match whether the flow query is streaming or batch".
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2719,6 +2719,33 @@
],
"sqlState" : "42000"
},
+ "INVALID_FLOW_RELATION_TYPE" : {
Review Comment:
Any particular reason to use "relation" instead of "query"? I think the latter is more common in docs.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -55,6 +55,75 @@ trait GraphValidations extends Logging {
multiQueryTables
}
+ protected[graph] def validateFlowStreamingness(): Unit = {
+ flowsTo.foreach { case (destTableIdentifier, flows) =>
+ val destTableOpt = table.get(destTableIdentifier)
+
+ // If the destination identifier does not correspond to a table, it must be a view.
+ val destViewOpt = destTableOpt.fold(view.get(destTableIdentifier))(_ => None)
+
+ flows.foreach {
Review Comment:
Nitpick you can feel free to ignore: this is pretty heavily nested. Could remove a level of nesting with:
```
val resolvedFlows: Seq[ResolvedFlow] = flows.collect { case rf: ResolvedFlow => rf }
```
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2719,6 +2719,33 @@
],
"sqlState" : "42000"
},
+ "INVALID_FLOW_RELATION_TYPE" : {
+ "message" : [
+ "Flow <flowIdentifier> returns an invalid relation type."
+ ],
+ "subClass" : {
+ "FOR_MATERIALIZED_VIEW" : {
Review Comment:
Thoughts on being more descriptive with these like `STREAMING_RELATION_FOR_MATERIALIZED_VIEW` / `BATCH_RELATION_FOR_STREAMING_TABLE`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SDP] Validate streaming-ness of DFs returned by SDP table and standalone flow definitions [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #51208:
URL: https://github.com/apache/spark/pull/51208#issuecomment-2994706154
Let's file a JIRA
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org