[FLINK-40298][table] Opt-in name-based state schema evolution for RowData

[weiqingy]

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:

Step	Sub-task	Scope
PR-1	FLINK-40296	Object-level migrate hook on TypeSerializerSnapshot
PR-2	FLINK-40297	Route TTL-aware value migration through the hook
PR-3 (this PR)	FLINK-40298	Opt-in name-based schema evolution for RowData
PR-4	FLINK-40299	End-to-end state migration coverage on RocksDB

Each PR depends on the one before it. 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.

Because this branch is stacked, the diff here also contains the PR-1 and PR-2 commits. For the isolated diff of this step alone, weiqingy#12 has the same head branch based on PR-2 instead of master. It is a reference only and is not for merge.

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.

Extending coverage to ListState<RowData> and MapState<K, RowData>, and propagating migration through composite serializers, is 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 @Internal KeyedStateBackend 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 @Internal default method, and KeyedStateBackend gains an @Internal default 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, is tracked separately and will follow.

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?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5)

---

[flinkbot]

CI report:

• ef27f14 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build