ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4194117506
##########
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:
To make sure I'm understanding correctly:
- we're saying we will not try to reconcile the offset shuffle partitions
(if they are present) with the state metadata value. If the query fails on its
own with the FileNotFoundException it will be up to the user to fix the
checkpoint
- in cases where the offset is missing shuffle partitions key we will
populate it from state metadata if possible. Here we can still detect an issue
and fail the query with a more useful message if for some reason we couldn't
load it from state metadata
- in cases where the offset is missing the shuffle partitions key but there
are no stateful operators we can assume to populate it safely from session
config
Does that all sound right and lines up with your recommendation?
--
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]
