You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ivasilyev1 (via GitHub)" <gi...@apache.org> on 2026/10/03 00:55:54 UTC
[PR] [SPARK-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
ivasilyev1 opened a new pull request, #59221:
URL: https://github.com/apache/spark/pull/59221
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'common/utils/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
The changes in this PR:
- Ensure that both V1 and V2 offset logs emit rebound SQL conf
- Attempt to perform recovery of missing `spark.sql.shuffle.partitions` in V2 offset log by obtaining value from state metadata if possible
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
With current implementation the stateful shuffle partitions are not written out by OffsetSeqMetadataV2. If query is restarted and the session configured shuffle partitions changes, the query will fail.
The exact error scenario and steps to reproduce are described in: https://issues.apache.org/jira/browse/SPARK-59919
The missing config key also leads to possible crash scenario as in https://github.com/apache/spark/pull/58929, which was addressed separately in that code path.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
I think no
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Tested with the included unit tests in this PR as well as with manual reproduction in spark
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Generated-by: GPT 5.6 Luna
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
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: 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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4212920628
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
Review Comment:
Switched to this
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191363232
##########
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:
Sorry I got lost since it got back and forth multiple times... I just refreshed my head and reason about the thing by myself.
Here I see we are trying to restore the broken checkpoint probably impacted by this bug. But this is risky since we are now disrespectful of the config value in the offset log, not only the case of missing (which is unavoidable to find a fallback), but also the case the value exists. I understand the scenario how it happens and how this could help, but self-mitigation is open to produce another problem and we could be in a loop on fixing issue forever.
Also the self-mitigation isn't even possible when poisoned value of shuffle partitions is smaller than the actual shuffle partitions; the query won't fail and succeed to produce incorrect output and commit, destroying the self-mitigation.
So let's not make this too complicated - I'd rather propose accepting the limitation. If we want to be safer than this, let's just do the check of shuffle partitions vs state metadata and either fail the query (if we really believe there must be no case where they must be different in normal case) or leave warn log to inform users.
Once we get rid of this logic, I assume the test logic would be also simplified (less scenarios to concern).
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6056327261
> @ivasilyev1 Could you please help submitting a PR for branch-4.3/branch-4.2? There are merge conflicts on those branches. Thanks in advance!
@HeartSaVioR Sure no problem, just might not be until later today. (Unless you have done it yourself by then based on your other comment)
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173244815
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
Let's also have the test case for this to verify we fail the query instead of continuing with blindly taking the session config's shuffle partitions.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4189860023
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
Let's merge this first and incorporate this change to #58929. #58929 isn't really a hurry (unless it exposes a correctness issue).
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
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; it's a source of truth and 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: 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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4201858987
##########
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:
Updated
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191399573
##########
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 main concern for me is that the committed offset log cannot be overwritten (this is an invariant), so even if we found that the value is incorrect (based on state metadata), we will leave the poisoned offset log as it is while the batch has run with different setting than the offset log says. It'd be safer to require human's intervention, unfortunately.
(Users need to remove the poisoned offset log file which is bound to uncommitted batch manually, and restart the query. That would be able to trigger the mitigation logic.)
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4189852879
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
+ 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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
Review Comment:
Just another case of triggering the same thing:
StateMetadataPartitionReader generally allows the state metadata file to be absent, but this recovery path requires it to determine the original state-store partition count. Allowing the missing file to pass through falls back to the session configuration and recreates the same correctness risk.
We should fail the query and add a regression test covering missing state metadata.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5969127744
> **AI-assisted review disclaimer:** This review was produced by an LLM-based coding agent. The
> reproducer and conclusions below were verified by executing the focused test against the stated
> PR commit.
I think there is still a recovery hole for a checkpoint that has already gone through one failed
restart on an affected Spark version.
The sequence is:
1. Batch `N` is committed with five-partition state, but its V2 offset metadata has no
`spark.sql.shuffle.partitions` entry.
2. The query is restarted with `spark.sql.shuffle.partitions=10`. Before loading state, the old
code writes offset `N + 1` with `10` copied into its metadata.
3. Loading the five-partition state fails, leaving offset `N + 1` uncommitted but present.
4. On the next restart, this PR sees that the latest offset contains the key and skips recovery.
It therefore tries ten partitions again and fails in exactly the same way.
5. There should also be another restart after successfully re-executing `N + 1`: re-execution
commits the existing offset rather than rewriting it, so recovery must not leave the checkpoint
poisoned for the following restart.
I reproduced this against `2bcb0ffedc52ac76ed87d13a7b070f6920adf053`. The focused test fails with:
```text
CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS
.../state/0/5/1.delta does not exist
```
Temporarily bypassing the present-key short circuit in `recoverStatefulShufflePartitions` makes
both restarts in the test pass with five state stores. That was only a mutation check, not a
suggested implementation: the fix still needs to preserve legitimate repartition semantics.
The `STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART=false` override below only isolates this
case because #58929 has not landed in this PR's base. Without it, the separate SPARK-59662
`None.get` failure happens before this recovery path runs. When testing on top of #58929, the
override can be removed.
<details>
<summary>Regression test diff (modifies the existing recovery test)</summary>
```diff
@@
- test("VERSION_2 recovers state-store partitions when the offset conf is missing") {
+ test("VERSION_2 recovers state-store partitions after a failed legacy restart") {
@@
withSQLConf(
SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
@@
assert(!offsetLog.get(latestBatchId).get.metadataOpt.get.conf
.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+ // A restart on an affected Spark version copies 10 into a new offset before failing to load
+ // the five-partition state, leaving that offset uncommitted.
+ val poisonedEndOffset = inputData.addData((2, 0), (3, 0))
+ val poisonedOffset = offsetWithoutShufflePartitions.copy(
+ offsetsMap = offsetWithoutShufflePartitions.offsetsMap.map {
+ case (sourceId, _) => sourceId -> Some(poisonedEndOffset)
+ },
+ metadata = offsetWithoutShufflePartitions.metadata.copy(
+ conf = offsetWithoutShufflePartitions.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, "10")))
+ assert(offsetLog.add(latestBatchId + 1, poisonedOffset))
+
val query2 = startQuery(shufflePartitions = 10)
try {
- inputData.addData((2, 0), (3, 0))
query2.processAllAvailable()
assert(query2.streamingQuery.lastExecution.numStateStores === 5)
} finally {
query2.stop()
}
+ // The recovered batch commits the existing poisoned offset rather than rewriting it.
+ // Verify that the following restart is also recoverable.
+ val query3 = startQuery(shufflePartitions = 10)
+ try {
+ inputData.addData((3, 0), (4, 0))
+ query3.processAllAvailable()
+ assert(query3.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query3.stop()
+ }
+
assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
```
</details>
Run with:
```bash
build/sbt 'sql/testOnly org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite -- -z "failed legacy restart"'
```
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173224334
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
I assume this only happens with broken checkpoint (mostly in state metadata), so it doesn't happen often.
But when it really happens, we must accept the fact that we have no way to know the true value of state shuffle partitions for this query, and fail the query. We shouldn't do swallowing the exception and falling back to use the session config. If the value of session config is smaller than the actual state shuffle partitions, we will end up with silent correctness issue, which is much worse than breaking query.
It's just that we need a good error message - the error message should say that the checkpoint is broken and they have to drop the checkpoint and reprocess, so nuance is very important; users have to take the action and cost because of Spark bug + some other issue (maybe they led it or another system error).
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173261588
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
If we fail to get the value from both offset log and state metadata, yes we do not have a way to know the true value of state shuffle partitions, and we must not pick the session config value because this is against the reason why we pin the config.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190613084
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
Review Comment:
committedBatchId < 0
This doesn't guarantee full safeness; in some extreme scenario, we could only have the checkpoint where offset logs are available but commit logs aren't. Even in that case, Spark does not reject the query to run, so we need to handle the case as well.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
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; it's a source of truth and 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. We do not care about the value in old offset log file.
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: 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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6048661061
Let me know if I missed any or misinterpreted
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4171774811
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
**Nit (P3):** This try/catch(NonFatal) is a deliberate graceful-degradation path: if the state-metadata read fails, recovery is skipped and the query still starts with the unmodified metadata. The new tests cover successful recovery, restart-preservation, and the empty-metadata case, but none exercises this fallback. Consider adding a test that points a legacy VERSION_2 checkpoint (with the offset conf removed, as in the recovery test) at an unreadable or corrupt state-metadata directory and asserts the query still starts with unchanged metadata, so a future change that lets the read throw instead of degrading is caught.
**Verification:**
- **Regression:** Reproduce a failed state-metadata read during recovery on a legacy VERSION_2 checkpoint missing the conf, and confirm the query still starts with unmodified metadata; the assertion must fail if the read exception is allowed to propagate instead of being caught.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173224334
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
I assume this only happens with broken checkpoint (mostly in state metadata), so it doesn't happen often.
But when it happens, we must accept the fact that we have no way to know the true value of state shuffle partitions for this query, and fail the query. We shouldn't do swallowing the exception and falling back to use the session config. If the value of session config is smaller than the actual state shuffle partitions, we will end up with silent correctness issue, which is much worse than breaking query.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173249831
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
Yes, I think the only way it would happen is if the state metadata is somehow corrupt or unreadable.
You're saying this should throw instead of just logging a warning and using available metadata correct?
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191363232
##########
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:
Sorry I got lost since it got back and forth multiple times... I just refreshed my head and reason about the thing by myself.
Here I see we are trying to restore the broken checkpoint probably impacted by this bug. But this is risky since we are now disrespectful of the config value in the offset log, not only the case of missing (which is unavoidable to find a fallback), but also the case the value exists. I understand the scenario how it happens and how this could help, but self-mitigation is open to produce another problem and we could be in a loop on fixing issue forever.
Also the self-mitigation isn't even possible when poisoned value of shuffle partitions is smaller than the actual shuffle partitions; the query won't fail and succeed to produce incorrect output and commit, destroying the self-mitigation.
So let's not make this too complicated - I'd rather propose accepting the limitation. If we want to be safer than this, let's just do the check of shuffle partitions vs state metadata and fail the query if they disagree.
Once we get rid of this logic, I assume the test logic would be also simplified (less scenarios to concern).
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4200239393
##########
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:
Thanks, I think that all makes sense. Will get this changed tonight
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191363232
##########
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:
Sorry I got lost since it got back and forth multiple times... I just refreshed my head and reason about the thing by myself.
Here I see we are trying to restore the broken checkpoint probably impacted by this bug. But this is risky since we are now disrespectful of the config value in the offset log, not only the case of missing (which is unavoidable to find a fallback), but also the case the value exists. I understand the scenario how it happens and how this could help, but self-mitigation is open to produce another problem and we could be in a loop on fixing issue forever.
Also the self-mitigation isn't even possible when poisoned value of shuffle partitions is smaller than the actual shuffle partitions; the query won't fail and succeed to produce incorrect output and commit, destroying the self-mitigation.
So let's not make this too complicated - I'd rather propose accepting the limitation. ~If we want to be safer than this, let's just do the check of shuffle partitions vs state metadata and fail the query if they disagree.~ (This isn't even safe in case of state repartition...)
Once we get rid of this logic, I assume the test logic would be also simplified (less scenarios to concern).
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
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: 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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4191519470
##########
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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ val stateMetadataBatchId = if (committedBatchId >= 0) {
+ committedBatchId
+ } else {
+ latestBatchId
+ }
+ val numPartitionsOpt = try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ stateMetadataBatchId)
+ val partitionsOpt = stateMetadataReader.stateStoreNumPartitions
+ // If the query was stateful and we could not get the number of partitions
+ // for any reason, abort the query.
+ if (partitionsOpt.isEmpty && containsStatefulOperator(analyzedPlan)) {
+ throw new IllegalStateException(
Review Comment:
We tend to associate the error to error class framework as much as possible rather than normal exception.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #59221: [SPARK-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding
URL: https://github.com/apache/spark/pull/59221
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5969242200
> > **AI-assisted review disclaimer:** This review was produced by an LLM-based coding agent. The
> > reproducer and conclusions below were verified by executing the focused test against the stated
> > PR commit.
>
> I think there is still a recovery hole for a checkpoint that has already gone through one failed restart on an affected Spark version.
>
> The sequence is:
>
> 1. Batch `N` is committed with five-partition state, but its V2 offset metadata has no
> `spark.sql.shuffle.partitions` entry.
> 2. The query is restarted with `spark.sql.shuffle.partitions=10`. Before loading state, the old
> code writes offset `N + 1` with `10` copied into its metadata.
> 3. Loading the five-partition state fails, leaving offset `N + 1` uncommitted but present.
> 4. On the next restart, this PR sees that the latest offset contains the key and skips recovery.
> It therefore tries ten partitions again and fails in exactly the same way.
> 5. There should also be another restart after successfully re-executing `N + 1`: re-execution
> commits the existing offset rather than rewriting it, so recovery must not leave the checkpoint
> poisoned for the following restart.
> ...
@HeartSaVioR Yes agreed, this is the scenario I mentioned yesterday on the JIRA ticket but my comment didn't describe the issue as well as this
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173248572
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -47,6 +51,18 @@ class OffsetSeqLogSuite extends SharedSparkSession {
assert(metadata.conf.get(SQLConf.ALLOW_EXCEPT_ON_STREAMING_DATAFRAME.key).contains("false"))
}
+ test("v1 and v2 metadata persist rebound stateful shuffle partitions") {
Review Comment:
nit: maybe add "SPARK-59919: " as prefix of the test case, since the ticket is worth mentioning. Apply to all test cases we are adding in this PR.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173248572
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -47,6 +51,18 @@ class OffsetSeqLogSuite extends SharedSparkSession {
assert(metadata.conf.get(SQLConf.ALLOW_EXCEPT_ON_STREAMING_DATAFRAME.key).contains("false"))
}
+ test("v1 and v2 metadata persist rebound stateful shuffle partitions") {
Review Comment:
nit: maybe add "SPARK-59919: " as prefix of the test case, since the ticket is worth mentioning.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4202312856
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,50 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ !containsStatefulOperator(analyzedPlan)) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ val stateMetadataBatchId = if (committedBatchId >= 0) committedBatchId else latestBatchId
+ val failedRecoveryMessage =
+ s"Failed to recover the state-store partition count from checkpoint " +
+ s"metadata at $stateCheckpointLocation. This can happen if the checkpoint " +
+ "was created using offset log format V2 in an unpatched Spark distribution " +
Review Comment:
nit: an unpatched <- this does not provide any context. If we want to be honest, we need to refer to the issue explicitly.
I'd suggest changing the sentence of "This can happen if the checkpoint ~ and state metadata was subsequently corrupted." to below:
This can happen if the checkpoint was created using offset log format V2 and state metadata was subsequently corrupted due to a bug. See SPARK-59919 for more details.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -47,6 +52,18 @@ class OffsetSeqLogSuite extends SharedSparkSession {
assert(metadata.conf.get(SQLConf.ALLOW_EXCEPT_ON_STREAMING_DATAFRAME.key).contains("false"))
}
+ test("[SPARK-59919] v1 and v2 metadata persist rebound stateful shuffle partitions") {
Review Comment:
nit: common practice is `SPARK-59919: ` rather than `[SPARK-59919]`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -825,9 +826,15 @@ class MicroBatchExecution(
// update offset metadata
nextOffsets.metadataOpt.foreach { metadata =>
- OffsetSeqMetadata.setSessionConf(metadata, sparkSessionToRunBatches.sessionState.conf)
+ val committedBatchId = commitLog.getLatestBatchId().getOrElse(-1L)
+ val metadataWithRecoveredPartitions = recoverStatefulShufflePartitions(
+ metadata, sparkSessionToRunBatches, latestBatchId, committedBatchId)
+ OffsetSeqMetadata.setSessionConf(
+ metadataWithRecoveredPartitions, sparkSessionToRunBatches.sessionState.conf)
execCtx.offsetSeqMetadata = OffsetSeqMetadata(
- metadata.batchWatermarkMs, metadata.batchTimestampMs, sparkSessionToRunBatches.conf)
+ metadataWithRecoveredPartitions.batchWatermarkMs,
+ metadataWithRecoveredPartitions.batchTimestampMs,
+ sparkSessionToRunBatches.conf)
watermarkTracker = WatermarkTracker(sparkSessionToRunBatches.conf, logicalPlan)
watermarkTracker.setWatermark(metadata.batchWatermarkMs)
Review Comment:
nit: since we build metadataWithRecoveredPartitions, probably good to be consistent, use `metadataWithRecoveredPartitions` rather than `metadata`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
Review Comment:
Maybe, could you give below a try and whether below can replace this validation? This validation is coupled with implementation detail for HDFS state store provider which is good to be avoided.
```
// The failure is expected because physical state exists for only five partitions.
// Planning ten stores proves the latest offset value won over state metadata.
assert(query2.streamingQuery.lastExecution.numStateStores === 10)
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
+ assert(stateStoreException.getCondition ===
+ "CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS")
+ assert(stateStoreException.getMessage.contains("does not exist"))
+ } finally {
+ query2.stop()
+ }
+ }
+ }
+
+ test("[SPARK-59919] VERSION_2 recovers state-store partitions when the commit log is missing") {
Review Comment:
For this test, we'd be better to explicitly set the checkpoint version to 1. Eventually we could flip the default and it would fail this test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -825,9 +826,15 @@ class MicroBatchExecution(
// update offset metadata
nextOffsets.metadataOpt.foreach { metadata =>
- OffsetSeqMetadata.setSessionConf(metadata, sparkSessionToRunBatches.sessionState.conf)
+ val committedBatchId = commitLog.getLatestBatchId().getOrElse(-1L)
Review Comment:
listing is a costly operation, especially against object store. We do the listing in L829 (here) and L845 (commitLog.getLatest()) - we can combine them and only do it once.
```
val latestCommittedBatch = commitLog.getLatest()
val committedBatchId = latestCommittedBatch.map(_._1).getOrElse(-1L)
```
Do this before L827, and remove L829 and replace L845 with `latestCommittedBatch match {`
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
Review Comment:
nit: shall we use checkError for the validation if it's available?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
+ assert(stateStoreException.getCondition ===
+ "CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS")
+ assert(stateStoreException.getMessage.contains("does not exist"))
+ } finally {
+ query2.stop()
+ }
+ }
+ }
Review Comment:
I don't see the test for fallback case, offset log is missing shuffle partitions (impacted) but the checkpoint has a readable state metadata which contains the shuffle partitions, and we continue with it. Please make sure we change the SQL config on restart to make the shuffle partitions in session config differs from shuffle partitions in state metadata (clearly, value in session config > value in state metadata, so that the query will fail if it doesn't read the state metadata).
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6058257358
PRs are submitted, no worries. I'll also review your next PR soon.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190652952
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
I realized #58929 isn't optional... I'll take a look at #58929 as well. Also the merge sequence will be simpler to go #58929 first...
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5963889917
@HeartSaVioR @micheal-o Please let me know if this meets the intent or if anything needs adjusting. I was particularly uncertain with the code paths around trying to load the partitions from state metadata if they were missing in the offset log
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173694228
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
Updated based on other thread t hat this should raise instead
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173704304
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
If https://github.com/apache/spark/pull/58929 is merged first I can rebase onto master and remove this
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190035031
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
+ 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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
Review Comment:
In this case then would we just check for the presence of the `state` directory, and based on this determine that it was a stateful query that we can no longer recover?
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190752951
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
Sorry, yes, it is not optional in that it can create a hard crash scenario too
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
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 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: 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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4194134874
##########
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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ val stateMetadataBatchId = if (committedBatchId >= 0) {
+ committedBatchId
+ } else {
+ latestBatchId
+ }
+ val numPartitionsOpt = try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ stateMetadataBatchId)
+ val partitionsOpt = stateMetadataReader.stateStoreNumPartitions
+ // If the query was stateful and we could not get the number of partitions
+ // for any reason, abort the query.
+ if (partitionsOpt.isEmpty && containsStatefulOperator(analyzedPlan)) {
+ throw new IllegalStateException(
Review Comment:
This was just an easy way to keep some of the logic a bit more condensed since this gets immediately caught and re raised with SparkException. I'll reorganize and remove while addressing the other comment
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4201877874
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
This particular comment is no longer relevant as far as removing the config key to disable offline repartition checks, however that PR still remains open and can cause a crash
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173224334
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
I assume this only happens with broken checkpoint (mostly in state metadata), so it doesn't happen often.
But when it really happens, we must accept the fact that we have no way to know the true value of state shuffle partitions for this query, and fail the query. We shouldn't do swallowing the exception and falling back to use the session config. If the value of session config is smaller than the actual state shuffle partitions, we will end up with silent correctness issue, which is much worse than breaking query.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5984115615
> Codex mentioned what I found manually and what Codex found are related but separate findings. So you may want to address both of them, with having each reproducer as test case.
@HeartSaVioR Think I addressed the remaining items with 06bc99c2969151acabe06a998eeb810231b7d03f. Let me know if I misunderstood or missed something. 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] [SPARK-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5965770224
On the recovery path you flagged: it looks correct to me. recoverStatefulShufflePartitions only triggers for a V2 offset log that is missing the stateful shuffle-partitions value, and that absence check resolves through the STATEFUL_SHUFFLE_PARTITIONS_INTERNAL -> SHUFFLE_PARTITIONS rebind to the same offset-log key that withStatefulShufflePartitions writes, so the guard and the writer agree. It reads the count from the committed batch's state metadata and falls back to the original metadata on any NonFatal error, and setSessionConf rebinds the value back to the session conf on restore. That matches the intended behavior.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5963889917","thread_id":"conversation:5963889917","verdict_sha256":"086acadc2f07e9fc032c379c9c79f6515423d82abb315be352fe5fa6ddd49004"} -->
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190287914
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
+ 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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
Review Comment:
If we see the stateful operator in the query but we don't see the shuffle partitions in offset log, and we fall back to find the state metadata and fail to read the shuffle partitions for that operator "in any reason", we must fail the query. Please focus on the part of "in any reason".
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190746231
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
+ 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.
+ // Re-execution commits that offset without rewriting it, so a missing value in the previous
+ // offset also triggers recovery on the following restart
+ val previousOffsetMissingStatefulShufflePartitions = latestBatchId > 0 &&
+ offsetLog.get(latestBatchId - 1).flatMap(_.metadataOpt)
+ .exists(previous => !previous.conf.contains(SQLConf.SHUFFLE_PARTITIONS.key))
+
+ if (hasStatefulShufflePartitions && !previousOffsetMissingStatefulShufflePartitions) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ val stateCheckpointLocation = new Path(checkpointFile("state")).getParent
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ stateCheckpointLocation.toString,
+ new SerializableConfiguration(
+ sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
Review Comment:
Updated so if there are stateful operators but we couldn't get the number of partitions it will fail
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,59 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ latestBatchId: Long,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 || committedBatchId < 0) {
Review Comment:
updated
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6009371370
My sincere apologize for the back-and-forth on review. I understand your frustration, my bad. I think LLM is pushing the code much more complicated just to achieve covering more cases, and I feel like we need to draw a line.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6053664336
Thanks! Merging to master/4.x/4.3/4.2.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5968804025
> On the recovery path you flagged: it looks correct to me. recoverStatefulShufflePartitions only triggers for a V2 offset log that is missing the stateful shuffle-partitions value, and that absence check resolves through the STATEFUL_SHUFFLE_PARTITIONS_INTERNAL -> SHUFFLE_PARTITIONS rebind to the same offset-log key that withStatefulShufflePartitions writes, so the guard and the writer agree. It reads the count from the committed batch's state metadata and falls back to the original metadata on any NonFatal error, and setSessionConf rebinds the value back to the session conf on restore. That matches the intended behavior.
@HyukjinKwon Thanks for this feedback. I will add a test case for the comment flagged above
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173224334
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
I assume this only happens with broken checkpoint (mostly in state metadata), so it doesn't happen often.
But when it happens, we must accept the fact that we have no way to know the true value of state shuffle partitions for this query, and fail the query. We shouldn't fall back to use the session config. If the value of session config is smaller than the actual state shuffle partitions, we will end up with silent correctness issue, which is much worse than breaking query.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4173261588
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -899,6 +906,42 @@ class MicroBatchExecution(
}
}
+ private def recoverStatefulShufflePartitions(
+ metadata: OffsetSeqMetadataBase,
+ sparkSessionToRunBatches: SparkSession,
+ committedBatchId: Long): OffsetSeqMetadataBase = {
+ if (metadata.version != OffsetSeqLog.VERSION_2 ||
+ OffsetSeqMetadata.readValueOpt(
+ metadata, SQLConf.STATEFUL_SHUFFLE_PARTITIONS_INTERNAL).isDefined ||
+ committedBatchId < 0) {
+ metadata
+ } else {
+ metadata match {
+ case v2: OffsetSeqMetadataV2 =>
+ try {
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ new Path(checkpointFile("state")).getParent.toString,
+ new SerializableConfiguration(sparkSessionToRunBatches.sessionState.newHadoopConf()),
+ committedBatchId)
+ stateMetadataReader.stateStoreNumPartitions.map { numPartitions =>
+ logWarning(log"Recovered state-store partition count " +
+ log"${MDC(CONFIG, SQLConf.SHUFFLE_PARTITIONS.key)}=" +
+ log"${MDC(NUM_PARTITIONS, numPartitions)} " +
+ log"from checkpoint state metadata because it was missing from the offset log")
+ OffsetSeqMetadata.withStatefulShufflePartitions(v2, numPartitions)
+ }.getOrElse(metadata)
+ } catch {
+ case NonFatal(e) =>
Review Comment:
If we fail to get the value from both offset log and state metadata, yes we do not have a way to know the true value of state shuffle partitions, and we must not pick the session config value because this violates why we pin the config.
--
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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-5969390669
Codex mentioned what I found manually and what Codex found are related but separate findings. So you may want to address both of them.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6035267122
> Almost there! Not a blocker level comment but several majors which are good to be addressed before merging.
Thanks! All make sense and will get them addressed
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4212925059
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +309,242 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString) {
+ aggregated.writeStream
+ .format("memory")
+ .queryName(queryName)
+ .outputMode("complete")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .start()
+ .asInstanceOf[StreamingQueryWrapper]
+ }
+ }
+
+ private def createStatefulCheckpoint(
+ checkpointDir: File,
+ queryName: String): (MemoryStream[(Int, Int)], DataFrame) = {
+ val inputData = MemoryStream[(Int, Int)]
+ val aggregated = inputData.toDS().groupBy("_1").count()
+ val query = startStatefulQuery(aggregated, checkpointDir, 5, queryName)
+ try {
+ inputData.addData((1, 0), (2, 0))
+ query.processAllAvailable()
+ inputData.addData((3, 0))
+ query.processAllAvailable()
+ assert(query.streamingQuery.lastExecution.numStateStores === 5)
+ } finally {
+ query.stop()
+ }
+ (inputData, aggregated)
+ }
+
+ private def assertStatefulPartitionRecoveryFails(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ queryName: String): Unit = {
+ val query = startStatefulQuery(aggregated, checkpointDir, 10, queryName)
+ try {
+ val exception = intercept[StreamingQueryException] {
+ query.processAllAvailable()
+ }
+ val message = exception.getCause.getMessage
+ assert(message.contains(
+ "Failed to recover the state-store partition count from checkpoint metadata"))
+ assert(message.contains("Delete the checkpoint and restart the query"))
+ } finally {
+ query.stop()
+ }
+ }
+
+ private def updateShufflePartitionsInOffset(
+ offsetLog: OffsetSeqLog,
+ checkpointDir: File,
+ batchId: Long,
+ numPartitions: Option[Int]): Unit = {
+ val offsetMap = offsetLog.get(batchId).get match {
+ case offset: OffsetMap => offset
+ case offset => fail(s"Expected a v2 offset, but found ${offset.getClass.getSimpleName}")
+ }
+ val conf = numPartitions match {
+ case Some(value) => offsetMap.metadata.conf.updated(
+ SQLConf.SHUFFLE_PARTITIONS.key, value.toString)
+ case None => offsetMap.metadata.conf - SQLConf.SHUFFLE_PARTITIONS.key
+ }
+ val updatedOffset = offsetMap.copy(metadata = offsetMap.metadata.copy(conf = conf))
+
+ val offsetFile = new Path(
+ new Path(checkpointDir.getAbsolutePath, "offsets"), batchId.toString)
+ val fileSystem = offsetFile.getFileSystem(spark.sessionState.newHadoopConf())
+ assert(fileSystem.delete(offsetFile, false))
+ assert(offsetLog.add(batchId, updatedOffset))
+ assert(offsetLog.get(batchId).get.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key) === numPartitions.map(_.toString))
+ }
+
+ test("[SPARK-59919] VERSION_2 uses offset partitions when state metadata differs") {
+ withTempDir { checkpointDir =>
+ val (inputData, aggregated) = createStatefulCheckpoint(
+ checkpointDir, "offsetlog_v2_stateful_restart_test")
+
+ val offsetLog = new OffsetSeqLog(spark, s"${checkpointDir.getAbsolutePath}/offsets")
+ val metadata = offsetLog.getLatest().get._2.metadataOpt.get
+ assert(metadata.version === OffsetSeqLog.VERSION_2)
+ assert(metadata.conf.get(SQLConf.SHUFFLE_PARTITIONS.key).contains("5"))
+ val latestBatchId = offsetLog.getLatestBatchId().get
+ assert(latestBatchId > 0L)
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId - 1, None)
+
+ // Keep the state metadata at 5 and change the latest offset to 10. The resulting state-store
+ // load failure proves the latest offset value was used instead of state metadata.
+ updateShufflePartitionsInOffset(offsetLog, checkpointDir, latestBatchId, Some(10))
+ assert(offsetLog.getLatest().get._2.metadataOpt.get.conf
+ .get(SQLConf.SHUFFLE_PARTITIONS.key).contains("10"))
+ val stateMetadataReader = new StateMetadataPartitionReader(
+ checkpointDir.getAbsolutePath,
+ new SerializableConfiguration(spark.sessionState.newHadoopConf()),
+ latestBatchId)
+ assert(stateMetadataReader.stateStoreNumPartitions.contains(5))
+
+ val query2 = startStatefulQuery(
+ aggregated, checkpointDir, shufflePartitions = 10,
+ queryName = "offsetlog_v2_stateful_restart_test")
+ try {
+ inputData.addData((2, 0), (3, 0))
+ val exception = intercept[StreamingQueryException] {
+ query2.processAllAvailable()
+ }
+ val stateStoreException = exception.getCause.asInstanceOf[SparkException]
+ assert(stateStoreException.getCondition ===
+ "CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS")
+ assert(stateStoreException.getMessage.contains("does not exist"))
+ } finally {
+ query2.stop()
+ }
+ }
+ }
+
+ test("[SPARK-59919] VERSION_2 recovers state-store partitions when the commit log is missing") {
Review Comment:
Switched this + the other stateful tests to use v1 format to be explicit and not fail unexpectedly in the future
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6014231346
> My sincere apologize for the back-and-forth on review. I understand your frustration, my bad.
>
> I think LLM is pushing the code much more complicated and even break the invariant (e.g. source of truth) just to achieve covering more cases to not break, and I feel like we need to step back and draw a line.
>
> Unfortunately this would delegate some work on mitigation over breakage to users; I hope we could document the issue and the mitigation to guide users for that manual mitigation. It's bad, but I'd accept the trade-off.
>
> Rationale: After pushing this way, LLM found another blocker/major based on what we have addressed, and I feel like it will continue as long as we just try to self-mitigate with leaving the broken part of it.
@HeartSaVioR 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-59919] OffsetSeqMetadataV2 missing shuffle partitions [spark]
Posted by "ivasilyev1 (via GitHub)" <gi...@apache.org>.
ivasilyev1 commented on code in PR #59221:
URL: https://github.com/apache/spark/pull/59221#discussion_r4190040251
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/OffsetSeqLogSuite.scala:
##########
@@ -292,6 +308,203 @@ class OffsetSeqLogSuite extends SharedSparkSession {
}
}
+ private def startStatefulQuery(
+ aggregated: DataFrame,
+ checkpointDir: File,
+ shufflePartitions: Int,
+ queryName: String,
+ checkUnfinishedRepartition: Boolean = true): StreamingQueryWrapper = {
+ withSQLConf(
+ SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
+ SQLConf.SHUFFLE_PARTITIONS.key -> shufflePartitions.toString,
+ SQLConf.STREAMING_CHECK_UNFINISHED_REPARTITION_ON_RESTART.key ->
Review Comment:
No hurry, I just assumed that one would get merged first as it is simpler
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6054054780
Hmm let me just give a try to have backport PRs to save time.
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6053706651
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/5d17442dec43f6dbea12973f9da18197b0e8d27c
- merged into branch-4.x https://github.com/apache/spark/commit/320f23696c2c681f3bcebcfee585004a9ce1d0ee
*Posted by `merge_spark_pr.py`*
--
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-59919][SS] Fix OffsetSeqMetadataV2 missing shuffle partitions on rebinding [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #59221:
URL: https://github.com/apache/spark/pull/59221#issuecomment-6053721149
@ivasilyev1
Could you please help submitting a PR for branch-4.3/branch-4.2? There are merge conflicts on those branches. Thanks in advance!
--
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