weiqingy opened a new pull request, #28973: URL: https://github.com/apache/flink/pull/28973
This is the third PR of the [FLIP-527](https://cwiki.apache.org/confluence/spaces/FLINK/pages/353601981/FLIP-527+State+Schema+Evolution+for+RowData) implementation, split into a stack of small, independently reviewable PRs under the umbrella issue [FLINK-37732](https://issues.apache.org/jira/browse/FLINK-37732). Landing order: | Step | Sub-task | Scope | |---|---|---| | PR-1 | [FLINK-40296](https://issues.apache.org/jira/browse/FLINK-40296) | Object-level `migrate` hook on `TypeSerializerSnapshot` | | PR-2 | [FLINK-40297](https://issues.apache.org/jira/browse/FLINK-40297) | Route TTL-aware value migration through the hook | | **PR-3 (this PR)** | [FLINK-40298](https://issues.apache.org/jira/browse/FLINK-40298) | Opt-in name-based schema evolution for `RowData` | | PR-4 | [FLINK-40299](https://issues.apache.org/jira/browse/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. ## 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? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Opus 5) --- -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
