HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4194964639
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,75 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2) {
+ metadata
+ } else {
+ val hasStatefulShufflePartitions = OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined
+ // In old checkpoints, this config may first appear in an offset written
by a failed restart.
Review Comment:
The baseline is simple, we consider the offset log as "source of truth" - we
do not ignore the value of shuffle partitions we see from offset log; we do not
try to figure out whether the value is poisoned or not. If the value is
available, it is a valid value. It only matters for the offset log to be
restored; based on the logic, we will only use the offset log file for the
latest batch to know the value.
In any reason we do not see the value of shuffle partitions from offset log
(like the current bug), we try to get the value from state metadata. If we fail
to read the value in "any" reason, we fail the query.
For sure, we do not fall back to read the state metadata for stateless query
- 1) we don't need to override the value of pinned config from offset log to
session config 2) state metadata won't be available anyway.
--
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]
