[
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: [email protected]
For additional commands, e-mail: [email protected]
