+1 for the proposal. Best, Shengkai
Weiqing Yang <[email protected]> 于2026年7月9日周四 14:14写道： > 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 >> > >> >
