Jungtaek Lim created SPARK-59919:
------------------------------------
Summary: 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
# 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]
