Why does my streaming query fail to restart after a code change?

Summary

• A restarted Structured Streaming query resumes from its checkpoint, which records the offsets each source has processed and the state of every stateful operator. If the new code no longer matches what the checkpoint describes, the restart fails.
• The Spark programming guide spells out the rules: filters and source rate limits can change between restarts, but the number or type of input sources and any stateful operation can't.
• State mismatches surface as STATE_STORE_KEY_SCHEMA_NOT_COMPATIBLE or STATE_STORE_VALUE_SCHEMA_NOT_COMPATIBLE. Disabling the schema check doesn't migrate the stored state.
• Fix it by reverting to a compatible change, or by running the query on a new checkpoint location with a deliberate starting position.
• For planned evolution, transformWithState supports value-side state schema changes with Avro encoding, and Spark 4.2.0 adds experimental source naming with DataStreamReader.name().

Why does my streaming query fail to restart after a code change?

Because a restarted streaming query isn't a new query. Structured Streaming resumes from the checkpoint, which stores the range of offsets processed in each trigger and the running state of stateful operators, and it assumes your new code describes the same sources and the same state schema. When it doesn't, the Spark programming guide is direct: the restarted query "is likely to fail with unpredictable errors." Two categories of change carry a flat "not allowed" in the guide: a change in the number or type of input sources, and any change to a stateful operation.

Which changes the checkpoint tolerates

The guide's "Recovery Semantics after Changes in a Streaming Query" section defines the rules. "Allowed" means Spark permits the change, though whether its effect is well-defined depends on the query.

Change between restarts	Rule	Example from the guide
Number or type of input sources	Not allowed	Adding a second source, or reading from a different source
Source rate limits	Allowed	Adding maxOffsetsPerTrigger to a Kafka source
Subscribed topics or files	Generally not allowed; results are unpredictable	subscribe changed from topic to newTopic
Output sink type	Case by case	File sink to Kafka sink is allowed; Kafka sink to file sink is not
Output sink parameters	Depends on the sink	A file sink's output directory can't change; a Kafka sink's output topic can
Filters and projections	Some cases allowed	Adding or removing filters is allowed; a projection with a different output schema only if the sink accepts that schema change
Stateful operations	Not allowed	New grouping keys or aggregates, different deduplication columns, changed stream-stream join columns or join type, or a new state schema or timeout type in mapGroupsWithState / flatMapGroupsWithState

Spark restores state by schema, so in the guide's words, "any changes (that is, additions, deletions, or schema modifications) to the stateful operations of a streaming query are not allowed between restarts."

Reading the error

A changed state schema stops the restart with one of two errors, both SQLSTATE XXKST:

• STATE_STORE_KEY_SCHEMA_NOT_COMPATIBLE: the provided key schema doesn't match the stored state key schema, typically after a change to grouping or deduplication columns.
• STATE_STORE_VALUE_SCHEMA_NOT_COMPATIBLE: the provided value schema doesn't match, for example after changing the aggregates.

Both messages print the existing and new schemas, mention spark.sql.streaming.stateStore.stateSchemaCheck, and warn that running with an incompatible schema "could cause non-deterministic behavior." Turning the check off doesn't convert the stored state; it only stops Spark from telling you it's wrong.
To see what the checkpoint holds before you decide, the State Data Source, marked experimental, reads it as a batch DataFrame:

ckpt = "/checkpoints/orders_agg"
spark.read.format("state-metadata").load(ckpt).show(truncate=False)
spark.read.format("statestore").load(ckpt).printSchema()

state-metadata lists the checkpoint's stateful operators (operatorId, operatorName, stateStoreName), and statestore reads the state itself.

How to fix it

1. Revert if you can. Redeploy a compatible version against the same checkpoint and the query resumes where it stopped.
2. Start fresh when the change is intentional. Point checkpointLocation at a new path and the query starts as a new query. For Kafka, startingOffsets "only applies when a new query is started," and its streaming default is latest. Choose earliest if the topic still retains the history you need to rebuild state, and make sure the sink can take rows it has already received.
3. In Spark Declarative Pipelines, spark-pipelines run --full-refresh <dataset> does that reset for you. It clears the dataset's data and checkpoints and recomputes it from scratch.

Designing queries that can evolve

transformWithState, the Arbitrary State API v2 introduced in Spark 4.0, documents state schema evolution on the value side. You can add state variables, or remove one by calling deleteIfExists in the processor's init method. Within a variable you can add, remove, reorder or widen fields once spark.sql.streaming.stateStore.encodingFormat is set to avro. Renaming fields, narrowing types and any key-side change remain unsupported.
Spark 4.2.0 adds DataStreamReader.name(), an experimental API that stores a stable source name in checkpoint metadata; SQL gets a matching IDENTIFIED BY clause. The release notes say named sources can be added, removed and reordered without breaking checkpoints:

orders = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "orders")
    .name("orders")
    .load()
)

How Databricks approaches query changes

The same recovery rules apply on Databricks. Databricks Runtime 18.2 and above documents source evolution: with spark.sql.streaming.queryEvolution.enableSourceEvolution set to true, every source must be named with .name() and names must be unique. It requires a fresh checkpoint and can't be turned off for that checkpoint afterward, so name sources when a query is first deployed. In Lakeflow Pipelines, a full refresh clears a streaming table's data and its flows' checkpoints and reprocesses the source, which can drop records if the source no longer holds them.

FAQs

Can I change the Kafka topics a query subscribes to?

Changes to subscribed topics are generally not allowed because the results are unpredictable. Read the new topic in its own query, or move the existing query to a new checkpoint.

Does deleting the checkpoint fix the error?

It removes the error by removing the history: the next run begins fresh, without the old offsets or state. A new checkpoint path does the same while keeping the old one for rollback.