[jira] [Comment Edited] (SPARK-59919) Offset log format v2 does not preserve stateful shuffle partitions across query restarts

["Jungtaek Lim (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-59919?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18122139#comment-18122139 ] 

Jungtaek Lim edited comment on SPARK-59919 at 10/3/26 12:24 AM:
----------------------------------------------------------------

[~ivasil] 

Let's fix the issue first (an invariant is broken), and then we can get back to that question.

IIUC (and folk who actively works on this subject confirmed it), at this point, SQL config (pinned) is a source of truth. We generally expect pinned SQL config and state metadata won't be different, but if it happens we may want to look into the case.

IMO, it is good to eventually move the source of the truth to state metadata, but backward compatibility is always an issue e.g. what if the streaming query was from the version which did not have state metadata.


was (Author: kabhwan):
[~ivasil] 

Let's fix the issue first (an invariant is broken), and then we can get back to that question. Personally we eventually want to move to state metadata, but backward compatibility is always an issue e.g. what if the streaming query was from the version which did not have state metadata.

> 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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org