viirya commented on code in PR #58562:
URL: https://github.com/apache/spark/pull/58562#discussion_r3964395939
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala:
##########
@@ -282,8 +284,40 @@ class StreamingQueryManager private[sql] (
catalogAndIdent,
catalogTable)
+ // Unsupported-operation checks run during this initial analysis, before
the normal offset-log
+ // configuration restoration. Read the compatibility value early for
existing queries that
+ // contain a streaming EXCEPT so that upgrading does not change whether
the query is accepted.
+ def hasExceptWithStreamingLeft: Boolean = analyzedPlan.exists {
+ case Except(left, _, _) if left.isStreaming => true
+ case _ => false
+ }
+
+ val streamWriteAnalysisSession = trigger match {
+ case _: ContinuousTrigger => sparkSession
+ case _ if recoverFromCheckpointLocation &&
+ (!useTempCheckpointLocation ||
userSpecifiedCheckpointLocation.isDefined) &&
Review Comment:
Could we also account for `spark.sql.streaming.checkpointLocation` here?
`useTempCheckpointLocation` indicates that the sink allows a temporary
checkpoint, rather than that the query actually uses one.
For example, a `foreachBatch` query can use a persistent checkpoint
configured through the session, together with a fixed `queryName`, without
setting the writer's `checkpointLocation` option. In that case,
`useTempCheckpointLocation` is `true` and `userSpecifiedCheckpointLocation` is
`None`, so this condition skips restoring the compatibility setting. Restarting
an existing query after upgrading would then fail the new EXCEPT check, despite
having a valid checkpoint.
Could we include this configuration path and add a restart regression test
using the session setting plus `queryName`?
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/StreamingQueryManager.scala:
##########
@@ -282,8 +284,40 @@ class StreamingQueryManager private[sql] (
catalogAndIdent,
catalogTable)
+ // Unsupported-operation checks run during this initial analysis, before
the normal offset-log
+ // configuration restoration. Read the compatibility value early for
existing queries that
+ // contain a streaming EXCEPT so that upgrading does not change whether
the query is accepted.
+ def hasExceptWithStreamingLeft: Boolean = analyzedPlan.exists {
+ case Except(left, _, _) if left.isStreaming => true
+ case _ => false
+ }
+
+ val streamWriteAnalysisSession = trigger match {
+ case _: ContinuousTrigger => sparkSession
+ case _ if recoverFromCheckpointLocation &&
+ (!useTempCheckpointLocation ||
userSpecifiedCheckpointLocation.isDefined) &&
+ hasExceptWithStreamingLeft =>
+ val (resolvedCheckpointRoot, _) =
+ ResolveWriteToStream.resolveCheckpointLocation(dataStreamWritePlan)
Review Comment:
Should this checkpoint resolution run inside `sparkSession.withActive`?
`ResolveWriteToStream` reads its configuration through `SQLConf.get`. The
existing resolution during analysis runs under the owning session, but this new
call happens outside that scope.
If the DataFrame belongs to session B while session A is active, this can
resolve the checkpoint using A's configuration. For a sink requiring a
persistent checkpoint, with the location configured only in B, this could
incorrectly report a missing checkpoint location or read compatibility metadata
from a different location.
Could we bind the resolution to the owning session and cover this with a
regression test?
It would also be preferable to resolve the location once and reuse it during
analysis. This method creates the checkpoint directory, and with a
session-level checkpoint root but no query name, the two calls can generate
different UUID paths, leaving an unused directory.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
