weiqingy opened a new pull request, #28678: URL: https://github.com/apache/flink/pull/28678
## What is the purpose of the change This pull request adds an opt-in, name-based **state schema evolution** path for `RowData` keyed state, implementing [FLIP-527](https://cwiki.apache.org/confluence/display/FLINK/FLIP-527) (JIRA: FLINK-37732). Today `RowDataSerializer.resolveSchemaCompatibility` returns `incompatible()` on any field-level change, so a Flink SQL job whose keyed state holds a `RowData` whose 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-`ROW` recursion), 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, so `resolveSchemaCompatibility` still returns `incompatible()` 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 - *(core)* Add an object-level `migrate(TypeSerializerSnapshot<T> old, T value)` default method to `TypeSerializerSnapshot` (default = identity; only `RowData` overrides it), so a serializer can remap a value written by a prior, compatible schema. - *(table)* Add `String[] fieldNames` to `RowDataSerializer` and bump `RowDataSerializerSnapshot` `V3 → V4` (append field names; V4 reads V3 with names absent → position-based behavior, preserving old savepoints). - *(table)* Implement name-based `resolveSchemaCompatibility` (returns `compatibleAfterMigration()` only when both snapshots carry names and the change is backward-compatible: nullable additions, reorder, nested `ROW`; rejects removals, renames, type changes, non-nullable additions) plus the `migrate` remap into a `GenericRowData` of the new arity (preserves `RowKind`, nulls for added fields, recurses nested `ROW`). - *(table)* Add the `table.exec.state.schema-evolution.enabled` option (default `false`). - *(runtime)* Drive the object-level `migrate` through `TtlAwareSerializer.migrateValueFromPriorSerializer` (polymorphic; preserves the TTL timestamp; no `flink-table` dependency), so TTL and non-TTL value state share one migration path. - *(table/core)* Gate field-name population on the option through a single global point: a new `@PublicEvolving SerializerConfig.isStateSchemaEvolutionEnabled()` (default `false`) that `InternalTypeInfo.createSerializer` reads to build a name-retaining `RowData` serializer only when the option is on. This covers every operator whose keyed-state serializer is built from an `InternalTypeInfo` descriptor (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: - Serializer-level evolution and rejection cases (nullable-add at various positions, reorder, nested `ROW`, combinations; rejections for removal, rename, type change, non-nullable add) in `flink-table-type-utils`. - `V3 → V4` snapshot backward-read (a genuine V3-layout snapshot restores with `fieldNames` null and position-based behavior). - The opt-in flag transport/copy regression (`SerializerConfigImplTest`) — the flag survives `configure()`/`copy()` on the shared job `Configuration`. - A deterministic RocksDB-backed migration test driving the full stack (RocksDB restore → `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: - Dependencies (does it add or upgrade a dependency): **no** - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **yes** — a new `@PublicEvolving` default method `TypeSerializerSnapshot.migrate(...)`, a new `@PublicEvolving` default method `SerializerConfig.isStateSchemaEvolutionEnabled()`, and a new `@PublicEvolving` config option `table.exec.state.schema-evolution.enabled`. All are `default`/additive, so existing implementors remain source- and binary-compatible. - The serializers: **yes** — `RowDataSerializer` gains optional field names and its snapshot format is bumped `V3 → V4` (backward-compatible read of V3). - The runtime per-record code paths (performance sensitive): **no** — field names are metadata used only at compatibility resolution and restore-time migration; per-record serialization/deserialization is unchanged and `equals`/`hashCode` ignore names. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **yes** — savepoint/checkpoint state restore. The snapshot format bump reads old (V3) snapshots unchanged, and the new migration path only activates when the opt-in is enabled; with the option off, restore compatibility is identical to today. - The S3 file system connector: **no** ## Documentation - Does this pull request introduce a new feature? **yes** - If yes, how is the feature documented? **docs** (the `table.exec.state.schema-evolution.enabled` option 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 `InternalTypeInfo` descriptor. 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? - [X] Yes (Claude Code, Opus 4.8) Generated-by: Claude Code (Opus 4.8) -- 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]
