Hi all, I'd like to revive this discussion. Since the last round I've re-aligned the proposal with current master, folded in all of the feedback, mirrored the proposal to a cwiki FLIP page (as Shengkai suggested), and opened a draft PR so the design can be reviewed against working code.
Proposal doc (updated): link <https://docs.google.com/document/d/1WtAxp-jAVTLMOfWNldLCAoK137P0ZCMxR8hOZGcMxuc/edit> cwiki FLIP-527: link <https://cwiki.apache.org/confluence/spaces/FLINK/pages/353601981/FLIP-527+State+Schema+Evolution+for+RowData> Draft PR (implements the proposal; kept in draft until we converge): link <https://github.com/apache/flink/pull/28678> Here's what changed, mapped to the feedback: 1. Single object-level migration hook (Zakelly, Hangxiang). The design converged on one default method on TypeSerializerSnapshot: T migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value). It is invoked on the new snapshot and receives the old one (the oldSerializerSnapshot naming Hangxiang suggested), with an identity default. Because it works on the already-deserialized object rather than on raw bytes, the same method covers value, list-element, and map-value migration uniformly. This supersedes the earlier byte-level migrateState(in, out) form and removes the need for a separate migrateElement — which answers the migrateElement question you both raised (Zakelly, Hangxiang): the object hook handles list elements and map values with the same method, so no dedicated element hook is needed. It also addresses Hangxiang's point that a common interface shouldn't grow methods most implementations won't use, as they inherit the identity default and are entirely unaffected. 2. The SchemaEvolutionSerializer-on-the-compatibility-result alternative (Hangxiang). This is written up under Rejected Alternatives: TypeSerializerSchemaCompatibility is a result holder rather than an executor, and a migration is a distinct transform role (old serialized bytes -> new layout), so a default method on the snapshot is the smaller, more targeted change. 3. Field metadata (Shengkai). Adopted String[] fieldNames on RowDataSerializer instead of the full RowType. This approach is lightweight and exactly enough for name-based mapping. 4. The opt-in, and whether it is necessary (Shengkai, Gabor; and the question I left open last July). The feature is gated by table.exec.state.schema-evolution.enabled (default false), scoped under table.exec.state.* to signal it is Table/RowData-specific for now — which directly addresses Gabor's point that the config name should make clear it is RowData-specific rather than a generic state-evolution switch. On whether the opt-in is necessary: it is the fail-closed safety switch for exactly the case Shengkai raised, where a SQL change can silently shift operator-internal buffers (e.g. inserting SUM(d) before SUM(c)) even when field names appear to match. With the option off, serializers are built name-less and behavior is byte-for-byte as today; enabling it is a deliberate per-job confirmation that the change preserves state mapping. 5. Scope and a concrete example (Shengkai, Hongshun). The FLIP now leads with the primary case Shengkai steered toward — the SQL is unchanged and the input schema evolves backward-compatibly — with a worked end-to-end example: a Kafka fact stream joined with a CDC dimension whose nested profile ROW gains a nullable field. Because the join buffers the whole dimension row in keyed state, the evolved ROW actually reaches state (a leaf projection like metadata.userId would be column-pruned and would not), so the savepoint restore exercises exactly this feature. That also serves as the connector example Hongshun asked for. On Shengkai's broader concern (July 23) that the feature may not be accessible to users because few understand the SQL operator state structure: the reframing above is my attempt to address it — by leading with the SQL-unchanged case and a concrete join example rather than the operator-internal view. Shengkai, I'd especially welcome your read on whether this framing now makes the feature's applicability clear. Two boundaries I've made explicit this round: - RocksDB is the initial target backend; the ForSt sync backend can follow, and ForSt async is out of scope. - RowData nested below a composite serializer (List or Tuple, e.g. interval- and outer-join buffers) is out of scope for now and fails closed (rejected on restore, never mis-migrated); propagating migration through composite serializers is a planned follow-up. The proposal doc and cwiki have the full details, worked examples, the V3->V4 snapshot compatibility story, and rejected alternatives. Feedback is very welcome. If the direction looks good after this round, I'll start a VOTE. Thanks again for all the input, Weiqing On Tue, Aug 19, 2025 at 7:34 AM Gabor Somogyi <[email protected]> wrote: > Hi Weiqing, > > I've just read through the whole FLIP and +1 on the direction. > > I've a comment apart from the other pending items. Namely the > configuration is > `state.schema-evolution.enable` which implied to me that it's a generic > state evolution > feature but it's limited to Row data. Maybe we can mark that it's Row data > specific. > I'm pretty sure that we're going to add further types but not all. > > BR, > G > > On 2025/04/26 05:45:32 Weiqing Yang wrote: > > Hi all, > > > > I’d like to initiate a discussion about enhancing state schema evolution > > support for RowData in Flink. > > > > *Motivation* > > > > Flink applications frequently need to evolve their state schema as > business > > requirements change. Currently, when users update a Table API or SQL job > > with schema changes involving RowData types (particularly nested > > structures), they encounter serialization compatibility errors during > state > > restoration, causing job failures.The issue occurs because existing state > > migration mechanisms don't properly handle RowData types during schema > > evolution, preventing users from making backward-compatible changes like: > > > > - > > > > Adding nullable fields to existing structures > > - > > > > Reordering fields within a row while preserving field names > > - > > > > Evolving nested row structures > > > > This limitation impacts production applications using Flink's Table API, > as > > the RowData type is central to this interface. Users are forced to choose > > between maintaining outdated schemas or reprocessing all state data when > > schema changes are required. > > > > Here’s the proposal document: Link > > < > https://docs.google.com/document/d/1WtAxp-jAVTLMOfWNldLCAoK137P0ZCMxR8hOZGcMxuc/edit?tab=t.0 > > > > Your feedback and ideas are welcome to refine this feature. > > > > Thanks, > > Weiqing > > >
