[
https://issues.apache.org/jira/browse/SPARK-59919?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18122140#comment-18122140
]
B. Micheal Okutubo commented on SPARK-59919:
--------------------------------------------
Currently the offsetlog is the main one we use, but it would typically be the
same value as the metadata. In the future, we would move to using state
metadata as source of truth
> Offset log format v2 does not preserve stateful shuffle partitions across
> query restarts
> ----------------------------------------------------------------------------------------
>
> Key: SPARK-59919
> URL: https://issues.apache.org/jira/browse/SPARK-59919
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 4.2.0
> Reporter: Jungtaek Lim
> Priority: Blocker
>
> # Description
> Offset log format v2 does not persist the stateful shuffle partition count
> used by a streaming query.
> Consequently, when a stateful query is restarted with a different value of
> spark.sql.shuffle.partitions, Spark uses the current session value instead of
> the partition count recorded by the original execution. This changes the
> state-store partitioning and can make the checkpoint unrecoverable.
> This is independent of offline state repartitioning. A normal
> stop-and-restart reproduces it.
> Offset log v1 persists:
> spark.sql.shuffle.partitions =
> spark.sql.streaming.internal.stateStore.partitions
> through the rebindSQLConfsSessionToOffsetLog mapping. However,
> OffsetSeqMetadataV2.apply copies only relevantSQLConfs and omits this
> rebinding.
> During restart, IncrementalExecution.numStateStores cannot find the persisted
> value and falls back to the current session’s spark.sql.shuffle.partitions.
> # Reproduction
> 1. Enable offset log format v2.
> 2. Start a stateful streaming aggregation with spark.sql.shuffle.partitions=5.
> 3. Process and commit at least one batch.
> 4. Stop the query.
> 5. Set spark.sql.shuffle.partitions=10.
> 6. Restart the query from the same checkpoint and process another batch.
> For example:
> withSQLConf(
> SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
> SQLConf.SHUFFLE_PARTITIONS.key -> "5")
> { // Start a stateful aggregation, process data, and stop it. }
> withSQLConf(
> SQLConf.STREAMING_OFFSET_LOG_FORMAT_VERSION.key -> "2",
> SQLConf.SHUFFLE_PARTITIONS.key -> "10")
> { // Restart from the same checkpoint and process more data. }
> # Actual behavior
> Spark reports:
> Conf 'spark.sql.streaming.internal.stateStore.partitions' was not found
> in the offset log. Using existing session conf value '10'.
> The restarted physical plan uses:
> Exchange hashpartitioning(..., 10), REQUIRED_BY_STATEFUL_OPERATOR
> It then attempts to load state partitions that were never created by the
> original five-partition execution and fails, for example:
> CANNOT_LOAD_STATE_STORE.CANNOT_READ_DELTA_FILE_NOT_EXISTS
> .../state/0/6/1.delta does not exist
> # Expected behavior
> The query should retain the original five state-store partitions and restart
> successfully, regardless of the current session value of
> spark.sql.shuffle.partitions.
> # Scope and compatibility
> Offset log format v2 is not the default in Spark 4.2.0; the default remains
> v1. Therefore, default v1 queries are not affected.
> However, any query using or requiring offset log format v2 can produce a
> checkpoint that cannot be restarted safely when the session shuffle-partition
> configuration changes.
> This is a regression in the offset-log v2 implementation introduced under
> SPARK-54583 and is separate from the state-repartition-specific failure
> tracked by SPARK-59662.
> # Suggested resolution
> - Make v2 persist the same stateful shuffle-partition rebinding as v1.
> - Add a regression test covering a normal 5-partition → 10-partition session
> restart.
> - Define recovery behavior for existing v2 checkpoints that already lack the
> configuration. Fixing only future writes will not repair those checkpoints.
> Recovery may need to infer the partition count from committed operator/state
> metadata before falling back to
> the current session configuration.
> # Related issues
> - SPARK-54583
> - SPARK-59662
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
