[
https://issues.apache.org/jira/browse/SPARK-59919?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim updated SPARK-59919:
---------------------------------
Fix Version/s: 4.3.0
> 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
> Assignee: Ilya Vasilyev
> Priority: Blocker
> Labels: pull-request-available
> Fix For: 4.3.0, 4.4.0
>
>
> # 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]
