[FLINK-37732][table] FLIP-527: name-based state schema evolution for RowData#28678
Draft
weiqingy wants to merge 6 commits into
Draft
[FLINK-37732][table] FLIP-527: name-based state schema evolution for RowData#28678weiqingy wants to merge 6 commits into
weiqingy wants to merge 6 commits into
Conversation
…apshot Add a default migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value) method to TypeSerializerSnapshot. The default returns the value unchanged, so a value already deserialized with the prior serializer is re-serialized as-is. Serializers whose in-memory representation is coupled to the schema override it to transform the value into the new layout during restore. Consistent with resolveSchemaCompatibility, the hook is invoked on the new snapshot and receives the old snapshot as its argument. It is inert in this commit -- no serializer overrides it yet, so behavior is unchanged everywhere. Generated-by: Claude Code (Opus 4.8)
…3->V4 Add a nullable String[] fieldNames to RowDataSerializer and a recursive name-retaining factory withFieldNames(RowType) that retains field names at every nesting level. The existing constructors stay names-less, so all current callers (including state keys) are unchanged and the feature is inert: nothing populates field names yet. Bump RowDataSerializerSnapshot from V3 to V4. writeSnapshot appends a field-names block (a boolean, then one UTF per field when present) after the types section and before the nested-serializer section; readSnapshot reads it only when readVersion >= 4, so a V3 snapshot restores with fieldNames null. The names are carried through snapshotConfiguration() and restoreSerializer(). resolveSchemaCompatibility and migrate are unchanged here; equals/hashCode continue to ignore fieldNames so a feature-off restore is behaviour-identical to today. This is a forward snapshot-format bump: a V4 serializer restores a V3 snapshot, but an older V3-only engine cannot read a V4 snapshot. Generated-by: Claude Code (Opus 4.8)
…gration for RowData Teach RowDataSerializerSnapshot to evolve RowData state by matching fields on name. resolveSchemaCompatibility now has three branches: an identical-types fast path that preserves today's nested-composite behavior exactly; a name-based evolution path (engaged only when both snapshots carry field names) that returns compatibleAfterMigration when every added field is nullable, every old field survives by name, leaf field types are unchanged, and nested ROW fields recurse; and incompatible() otherwise (a differing layout with field names absent on either side preserves today's behavior). A nested ROW field is exempted from the leaf type-equality check and validated by recursion instead, so an evolved nested ROW is accepted while a nested incompatible change is still rejected. Override the object-level migrate hook to remap an old RowData into the new layout: fields placed by name, added and null fields become null, RowKind preserved, nested ROW values remapped recursively. The remap is object-in/object-out; the state backend performs deserialize/serialize. Generated-by: Claude Code (Opus 4.8)
…tion Add an opt-in boolean option table.exec.state.schema-evolution.enabled (default false) to ExecutionConfigOptions, mirroring the table.exec.state.ttl precedent, and its generated config-docs row. When enabled, RowData state-value serializers retain field names so backward-compatible schema changes (added nullable fields, reorder-by-name, nested ROW evolution) are migrated on state restore; when disabled the serializer behaves exactly as today and any field-type change is rejected as incompatible. The option is inert in this commit -- no operator reads it yet. Default false leaves existing jobs unaffected. Generated-by: Claude Code (Opus 4.8)
…re migrate path Rewrite TtlAwareSerializer.migrateValueFromPriorSerializer to migrate the deserialized state value to the current schema: unwrap the prior value to its bare form, apply the object-level migrate hook of the inner value serializer's snapshot, then re-wrap in a TtlValue. The migrate hook is identity for serializers that do not override it, so non-RowData and identical-schema state is byte-identical to before; the RowData serializer overrides it to remap fields on a backward-compatible schema change. Dispatch is polymorphic through the flink-core TypeSerializerSnapshot.migrate hook, so flink-runtime gains no dependency on flink-table. When both sides are TTL-enabled the prior last-access timestamp is preserved, so migration does not reset state expiry. The three RocksDB state classes and RocksDBKeyedStateBackend are unchanged: list-element and map-value migration already funnel through this method. Generated-by: Claude Code (Opus 4.8)
…e.exec.state.schema-evolution.enabled Populate RowData state value field names through a single global, opt-in gate instead of per-operator wiring. When table.exec.state.schema-evolution.enabled is set (default false), InternalTypeInfo.createSerializer builds a name-retaining RowData serializer (RowDataSerializer.withFieldNames); otherwise it returns today's name-less serializer. Because every Table operator whose keyed-state serializer is built from an InternalTypeInfo descriptor flows through this method, the option covers regular/interval/temporal joins, group aggregate, deduplicate, rank, and over-aggregate at once, with no per-operator code. The flag travels with no new transport: the planner merges the table config into the job Configuration that backs SerializerConfig, which is serialized into the JobGraph and read at runtime state registration. A new default method SerializerConfig.isStateSchemaEvolutionEnabled() (default false) exposes it; SerializerConfigImpl reads it from the backing Configuration and copies the key in configure() so copy()/configure() preserve it. flink-core mirrors the option key by string because it cannot depend on flink-table-api-java. The gate names key serializers as well as value serializers, which is safe on RocksDB because the backend rejects key-schema migration on every key path. RowDataSerializer .duplicate() carries field names through, because the state backend registers a duplicate of the serializer and name-based compatibility depends on the names being present. Operators that pre-build their state serializer (window, lookup, process-table, sink materializer), outer joins (composite-wrapped value), and backends other than RocksDB are not covered here. Generated-by: Claude Code (Opus 4.8)
Collaborator
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
What is the purpose of the change
This pull request adds an opt-in, name-based state schema evolution path for
RowDatakeyed state, implementing FLIP-527 (JIRA: FLINK-37732).Today
RowDataSerializer.resolveSchemaCompatibilityreturnsincompatible()on any field-level change, so a Flink SQL job whose keyed state holds aRowDatawhose schema evolves — for example an upstream Avro/CDC source adds a nullable column — cannot restore from a savepoint. This change lets such state migrate on restore by matching fields by name (nullable additions, reordering, and nested-ROWrecursion), rather than by position.The feature is opt-in and off by default (
table.exec.state.schema-evolution.enabled = false) and, in this first PR, scoped to the RocksDB state backend. The opt-in is enforced at serializer construction: when the option is off, RowData state serializers are built name-less exactly as today, soresolveSchemaCompatibilitystill returnsincompatible()and restore behavior is unchanged. The design fails closed — nothing migrates unless a user explicitly enables the feature.This PR is intended to accompany the FLIP-527 discussion and is opened as a draft; it is not intended to merge until the FLIP is approved.
Brief change log
migrate(TypeSerializerSnapshot<T> old, T value)default method toTypeSerializerSnapshot(default = identity; onlyRowDataoverrides it), so a serializer can remap a value written by a prior, compatible schema.String[] fieldNamestoRowDataSerializerand bumpRowDataSerializerSnapshotV3 → V4(append field names; V4 reads V3 with names absent → position-based behavior, preserving old savepoints).resolveSchemaCompatibility(returnscompatibleAfterMigration()only when both snapshots carry names and the change is backward-compatible: nullable additions, reorder, nestedROW; rejects removals, renames, type changes, non-nullable additions) plus themigrateremap into aGenericRowDataof the new arity (preservesRowKind, nulls for added fields, recurses nestedROW).table.exec.state.schema-evolution.enabledoption (defaultfalse).migratethroughTtlAwareSerializer.migrateValueFromPriorSerializer(polymorphic; preserves the TTL timestamp; noflink-tabledependency), so TTL and non-TTL value state share one migration path.@PublicEvolving SerializerConfig.isStateSchemaEvolutionEnabled()(defaultfalse) thatInternalTypeInfo.createSerializerreads to build a name-retainingRowDataserializer only when the option is on. This covers every operator whose keyed-state serializer is built from anInternalTypeInfodescriptor (regular/interval/temporal joins, group aggregate, deduplicate, rank, over-aggregate) in one place, with no per-operator code.Verifying this change
This change added tests and can be verified as follows:
ROW, combinations; rejections for removal, rename, type change, non-nullable add) inflink-table-type-utils.V3 → V4snapshot backward-read (a genuine V3-layout snapshot restores withfieldNamesnull and position-based behavior).SerializerConfigImplTest) — the flag survivesconfigure()/copy()on the shared jobConfiguration.TtlAwareSerializer.migrateValueFromPriorSerializer→RowDataSerializerSnapshot.migrate) for value state (TTL and non-TTL) and map value, plus a leaf type-change rejection.Does this pull request potentially affect one of the following parts:
@Public(Evolving): yes — a new@PublicEvolvingdefault methodTypeSerializerSnapshot.migrate(...), a new@PublicEvolvingdefault methodSerializerConfig.isStateSchemaEvolutionEnabled(), and a new@PublicEvolvingconfig optiontable.exec.state.schema-evolution.enabled. All aredefault/additive, so existing implementors remain source- and binary-compatible.RowDataSerializergains optional field names and its snapshot format is bumpedV3 → V4(backward-compatible read of V3).equals/hashCodeignore names.Documentation
table.exec.state.schema-evolution.enabledoption is generated into the configuration reference) and JavaDocs on the new public methods. A dedicated documentation page describing supported schema changes and the RocksDB scope is a planned follow-up.Scope note for reviewers: this first PR is RocksDB-scoped and covers operators whose keyed-state serializer is built from an
InternalTypeInfodescriptor. Operators that pre-build their state serializer (window, lookup, process-table, sink materializer), outer joins (composite-wrapped value), and other backends (ForSt, Heap) are intentionally out of scope and fail closed (their state does not evolve yet); each is tracked as a follow-up under FLINK-37732.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 4.8)