You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
[FLINK-40298][table] Opt-in name-based state schema evolution for RowData - #28973
This is the third PR of the FLIP-527 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-37732. Landing order:
End-to-end state migration coverage on RocksDB, and the user documentation
PR-1 and PR-2 have merged, so this branch is now a single commit on master and the diff below is this step alone. This is where the feature becomes observable: PR-1 and PR-2 were behavior-neutral, because the hook added in PR-1 defaults to returning its argument and this PR adds the first override.
What is the purpose of the change
Lets a RowData state value survive a backward-compatible schema change instead of failing the restore, behind a new opt-in option that defaults to off.
Compatibility becomes name based. When the field layouts differ but the difference is a supported backward-compatible change, RowDataSerializerSnapshot.resolveSchemaCompatibility returns compatibleAfterMigration() rather than incompatible(). Fields are matched by the names persisted by FLINK-40120. Added fields must be nullable; removed fields and changed leaf types are rejected; nested ROW fields are validated by recursion. The migrate override then remaps the row into the new layout, placing fields by name, null filling added fields, preserving the row kind, and recursing into nested ROW values.
The opt-in is carried as an in-memory, non-persisted flag on the serializer rather than as anything written into the snapshot, so nothing about the option reaches the snapshot format.
A snapshot written before field names were persisted carries no names. Rather than being rejected, it falls back to positional mapping, bounded to the one edit under which position is a stable identity: the old layout must be a prefix of the new one and every appended field must be nullable. Without names, an insertion in the middle is indistinguishable from a retype plus an append, and the two demand opposite migrations, so nothing broader can be admitted safely.
Why the flag is set where it is
The flag is set only on a serializer that is a state's own value serializer, and only when the keyed state backend actually performs object-level value migration. Both halves are load-bearing, and together they are what makes the feature fail closed.
The governing invariant is that a serializer may carry the flag only if some backend will actually invoke migrate on that exact serializer. Where compatibility reports compatibleAfterMigration and nothing calls migrate, the backend accepts the restore and later writes the old binary layout back through a serializer of the new arity, which corrupts the value silently rather than failing.
Two consequences follow. First, a RowData serializer nested below a composite serializer, as in interval and outer join buffers, is never a state's own value serializer, so it never carries the flag and its state is rejected on restore rather than migrated. Second, KeyedStateBackend gains an @Internal capability method defaulting to false, which only the RocksDB backend overrides. The heap backend accepts compatibleAfterMigration and relies on the next checkpoint rewriting already deserialized objects, which is sound for schema-free objects but not for RowData, whose restored value is a BinaryRowData view over the old bytes. Operator state, broadcast state and the batch backend never invoke the hook either. All of them therefore leave the flag off and reject the restore.
ListState<RowData> and MapState<K, RowData> are in FLIP scope. They are armed in PR-3b, which extends this same seam by exactly one structural level, matching what the RocksDB list and map states already unwrap before they migrate. Propagating migration through composite serializers is outside the FLIP and stays follow-up work.
Brief change log
Add table.exec.state.schema-evolution.enabled, default false, and expose it to the serializer layer through SerializerConfig
Add an @Internal marker interface so a keyed state path can hand a serializer the opt-in without flink-core knowing about RowData
Add an @InternalKeyedStateBackend capability, default false, overridden only by the RocksDB backend, and arm the serializer only on that path
Remove five redundant serializer pre-initializations in StreamingRuntimeContext, which duplicated what DefaultKeyedStateStore does immediately afterwards with an equivalent factory
Implement name-based compatibility, the bounded positional fallback, and the migrate override in RowDataSerializerSnapshot
Verifying this change
This change added tests and can be verified as follows:
Compatibility and migration matrix: nullable field added, NOT NULL field added, field dropped, fields reordered, leaf type changed, nested ROW evolved, ARRAY<ROW> changed
Same-typed reorder and rename: a reorder of two fields of the same type has an identical LogicalType array, so it is covered explicitly to confirm it is matched by name rather than accepted positionally
Positional fallback: a name-less prior snapshot migrates an appended nullable field, and is rejected for an appended NOT NULL field, a dropped field, and a field retyped mid-layout. An unchanged layout still resolves as compatible as is, so enabling the option on an older savepoint does not force a rewrite of state whose schema did not change.
Opt-in semantics: option off preserves today's behavior, the flag is not persisted across a snapshot write and read, and duplicate() carries it
Fail-closed: a state descriptor shaped like an interval join buffer does not arm the nested RowData serializer and an evolved schema is rejected, and a backend that does not perform object-level migration leaves the serializer unarmed
The option survives construction, copy() and configure()
Does this pull request potentially affect one of the following parts:
Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes. SerializerConfig is @PublicEvolving and gains an @Internaldefault method, and KeyedStateBackend gains an @Internaldefault method. Both are source and binary compatible and no existing implementor needs to change. A new table option is added.
The serializers: yes
The runtime per-record code paths (performance sensitive): no, the added work is on the restore path. With the option off, the only added cost is a boolean check during compatibility resolution.
Anything that affects deployment or recovery: yes, it changes which restores are accepted when the option is enabled. With the option off, restore behavior is unchanged.
The S3 file system connector: no
Documentation
Does this pull request introduce a new feature? yes
If yes, how is the feature documented? The generated execution config option page, plus JavaDocs. The prose documentation the FLIP calls for, covering which schema changes are supported versus rejected, lands in PR-4 alongside the end-to-end coverage that exercises the behavior it describes.
Notes for reviewers comparing this against the FLIP
Four points where the implementation is narrower than, or additional to, the approved design. Flagging them here rather than leaving them to be found.
A new @Internal capability on KeyedStateBackend. The FLIP does not describe one. It became necessary because the flag alone does not establish that anything will act on it: HeapKeyedStateBackend accepts compatibleAfterMigration and relies on the next checkpoint rewriting already-deserialized objects, which is correct for schema-free objects but not for RowData, whose restored value is a BinaryRowData view over the old bytes. Operator state, broadcast state and the batch backend accept it and never invoke the hook either. Rather than enumerate the backends that do not migrate, the capability states which one does.
The positional fallback is bounded. The FLIP authorises a fallback to position-based mapping for a name-less prior snapshot without constraining it. This restricts it to an append-only change: the old layout must be a prefix of the new one and every appended field must be nullable. Without names, an insertion in the middle is indistinguishable from a retype followed by an append, and the two require opposite migrations.
Key serializers are not armed. The FLIP says the flag is applied uniformly without distinguishing key from value serializers. Here it reaches value serializers only. The effect is the same, since RowData as a state key is out of scope either way.
The option is exposed through SerializerConfig. The FLIP notes that persisting field names unconditionally keeps the opt-in a table-layer concern and avoids plumbing configuration into the serialization layer. Reading the option still requires it to reach the layer that arms the serializer, so SerializerConfig gains an @Internal accessor mirroring the table option by key. Suggestions for a cleaner route are welcome.
Was generative AI tooling used to co-author this PR?
PR-1 and PR-2 have been merged, so this PR has been rebased onto master. The diff now only contains this step.
This is the step where the feature is actually turned on, so the part where I’d most like a second opinion is where the opt-in flag is allowed to be set, under “Why the flag is set where it is” above.
The rule is: a serializer should only carry the flag on a path that actually calls migrate. Everywhere else should leave the flag off and reject the restore.
@gyfora@RocMarshal, could you take a look when you get a chance? Thanks.
…Data
Make RowData state survive a backward-compatible schema change behind
table.exec.state.schema-evolution.enabled, default false.
Compatibility becomes name based when the flag is armed: added fields must
be nullable, removed fields and changed leaf types are rejected, and nested
ROW fields are validated by recursion. A prior snapshot carrying no field
names falls back to positional mapping bounded to an append-only change,
since without names an insertion is indistinguishable from a retype.
The flag is in-memory and never persisted. It is set only on a state's own
value serializer, and only on a keyed backend that performs object-level
value migration, so a serializer can carry it only where migrate is
actually invoked. A RowData serializer nested below a composite serializer,
the heap and batch backends, and operator and broadcast state all leave it
off and reject the restore rather than accepting a migration that nobody
performs.
Generated-by: Claude Code (Opus 5)
…cover direct backend arming
A nested ROW field pair skipped the exact type check so its contents could
be compared by the nested snapshot recursion. That recursion compares the
ROW's fields and never the ROW's own nullability, so narrowing a nullable
ROW to NOT NULL was accepted and migration then carried an existing null
straight into the non-null field. Settle nullability before delegating,
which matches how a leaf field is treated, since LogicalType#equals
compares nullability too.
Arming a descriptor registered directly through
AbstractKeyedStateBackend#getOrCreateKeyedState had no coverage: the
existing tests drive DefaultKeyedStateStore and StreamingRuntimeContext,
and their fake backend overrides getOrCreateKeyedState. Window and async
operators reach that method directly. Cover both capability values against
a real AbstractKeyedStateBackend, which the mock backend now reports
through an opt-in builder flag.
Generated-by: Claude Code (Opus 5)
…e when the option is off
Enabling or disabling TTL on restore turns an unarmed compatibleAsIs into
compatibleAfterMigration, so TtlAwareSerializer hands every value to
RowDataSerializerSnapshot#migrate even with state schema evolution off.
migrate remapped by name unconditionally: a same-typed rename, which the
option-off resolution matched by position, lost its value, and a same-typed
reorder swapped values. Return the value unchanged unless the snapshot is
armed, which restores the default hook's behavior for the option-off path.
Generated-by: GitHub Copilot CLI (Claude Opus 5.5)
…onfiguring when the row evolves
An armed evolution resolves to compatibleAfterMigration, and migrate then
rebuilds only the ROW spine: every other field keeps the bytes its old
serializer wrote. A RAW field whose Kryo registration order changed still
has an equal RawType, so it passed the type check, while its nested
snapshot asked for a reconfigured serializer. That verdict was collapsed
into compatibleAfterMigration, and the copied payload kept the old
registration ids under the new serializer.
The evolution paths now require every aligned non-ROW field to be
compatible as is, and apply the same rule recursively to nested ROW
fields, including nested rows whose own layout is unchanged. A
reconfiguration is rejected rather than masked by another field's
migration. The identical-layout path is unchanged.
Generated-by: GitHub Copilot CLI (Claude Opus 5.5)
Unsupported paths retain legacy compatibility behavior; they do not always reject schema changes. For example, an unarmed serializer accepts (a INT, b INT) → (b INT, a INT) positionally because hasSameLayoutAs ignores field names when evolution is disabled. The restore succeeds without moving values to their new named positions. Describe unsupported backends, pre-built serializers, and nested composite serializers as retaining legacy behavior, and warn that same-typed reorders are not detected. Regenerate the execution configuration documentation after updating this description.
… in the schema-evolution option
The description said that a RowData the option does not cover is
still rejected at restore and that other backends still fail. Where
the option does not apply, and when it is disabled, the restore keeps
matching fields by position: a change in field count or type fails,
but a reorder or rename among same-typed fields is not detected and
the values stay where they were. Say so, and note that state written
before field names were stored (Flink 2.3 or earlier) is matched by
position as well, except for appended nullable fields.
Generated-by: GitHub Copilot CLI (Claude Opus 5.5)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
github-actions
Bot
added
the
community-reviewed
This is the third PR of the FLIP-527 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-37732. Landing order:
migratehook onTypeSerializerSnapshot(merged)RowDataListState<RowData>andMapState<K, RowData>(#29449, draft)PR-1 and PR-2 have merged, so this branch is now a single commit on master and the diff below is this step alone. This is where the feature becomes observable: PR-1 and PR-2 were behavior-neutral, because the hook added in PR-1 defaults to returning its argument and this PR adds the first override.
What is the purpose of the change
Lets a
RowDatastate value survive a backward-compatible schema change instead of failing the restore, behind a new opt-in option that defaults to off.Compatibility becomes name based. When the field layouts differ but the difference is a supported backward-compatible change,
RowDataSerializerSnapshot.resolveSchemaCompatibilityreturnscompatibleAfterMigration()rather thanincompatible(). Fields are matched by the names persisted by FLINK-40120. Added fields must be nullable; removed fields and changed leaf types are rejected; nestedROWfields are validated by recursion. Themigrateoverride then remaps the row into the new layout, placing fields by name, null filling added fields, preserving the row kind, and recursing into nestedROWvalues.The opt-in is carried as an in-memory, non-persisted flag on the serializer rather than as anything written into the snapshot, so nothing about the option reaches the snapshot format.
A snapshot written before field names were persisted carries no names. Rather than being rejected, it falls back to positional mapping, bounded to the one edit under which position is a stable identity: the old layout must be a prefix of the new one and every appended field must be nullable. Without names, an insertion in the middle is indistinguishable from a retype plus an append, and the two demand opposite migrations, so nothing broader can be admitted safely.
Why the flag is set where it is
The flag is set only on a serializer that is a state's own value serializer, and only when the keyed state backend actually performs object-level value migration. Both halves are load-bearing, and together they are what makes the feature fail closed.
The governing invariant is that a serializer may carry the flag only if some backend will actually invoke
migrateon that exact serializer. Where compatibility reportscompatibleAfterMigrationand nothing callsmigrate, the backend accepts the restore and later writes the old binary layout back through a serializer of the new arity, which corrupts the value silently rather than failing.Two consequences follow. First, a
RowDataserializer nested below a composite serializer, as in interval and outer join buffers, is never a state's own value serializer, so it never carries the flag and its state is rejected on restore rather than migrated. Second,KeyedStateBackendgains an@Internalcapability method defaulting tofalse, which only the RocksDB backend overrides. The heap backend acceptscompatibleAfterMigrationand relies on the next checkpoint rewriting already deserialized objects, which is sound for schema-free objects but not forRowData, whose restored value is aBinaryRowDataview over the old bytes. Operator state, broadcast state and the batch backend never invoke the hook either. All of them therefore leave the flag off and reject the restore.ListState<RowData>andMapState<K, RowData>are in FLIP scope. They are armed in PR-3b, which extends this same seam by exactly one structural level, matching what the RocksDB list and map states already unwrap before they migrate. Propagating migration through composite serializers is outside the FLIP and stays follow-up work.Brief change log
table.exec.state.schema-evolution.enabled, defaultfalse, and expose it to the serializer layer throughSerializerConfig@Internalmarker interface so a keyed state path can hand a serializer the opt-in without flink-core knowing aboutRowData@InternalKeyedStateBackendcapability, defaultfalse, overridden only by the RocksDB backend, and arm the serializer only on that pathStreamingRuntimeContext, which duplicated whatDefaultKeyedStateStoredoes immediately afterwards with an equivalent factorymigrateoverride inRowDataSerializerSnapshotVerifying this change
This change added tests and can be verified as follows:
ROWevolved,ARRAY<ROW>changedLogicalTypearray, so it is covered explicitly to confirm it is matched by name rather than accepted positionallyduplicate()carries itRowDataserializer and an evolved schema is rejected, and a backend that does not perform object-level migration leaves the serializer unarmedcopy()andconfigure()Does this pull request potentially affect one of the following parts:
@Public(Evolving): yes.SerializerConfigis@PublicEvolvingand gains an@Internaldefaultmethod, andKeyedStateBackendgains an@Internaldefaultmethod. Both are source and binary compatible and no existing implementor needs to change. A new table option is added.Documentation
Notes for reviewers comparing this against the FLIP
Four points where the implementation is narrower than, or additional to, the approved design. Flagging them here rather than leaving them to be found.
@Internalcapability onKeyedStateBackend. The FLIP does not describe one. It became necessary because the flag alone does not establish that anything will act on it:HeapKeyedStateBackendacceptscompatibleAfterMigrationand relies on the next checkpoint rewriting already-deserialized objects, which is correct for schema-free objects but not forRowData, whose restored value is aBinaryRowDataview over the old bytes. Operator state, broadcast state and the batch backend accept it and never invoke the hook either. Rather than enumerate the backends that do not migrate, the capability states which one does.RowDataas a state key is out of scope either way.SerializerConfig. The FLIP notes that persisting field names unconditionally keeps the opt-in a table-layer concern and avoids plumbing configuration into the serialization layer. Reading the option still requires it to reach the layer that arms the serializer, soSerializerConfiggains an@Internalaccessor mirroring the table option by key. Suggestions for a cleaner route are welcome.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 5)