Hi Devs! Thank you all for the thoughtful comments, I have tried to incorporate most of the larger design additions and changes to the FLIP document. I have also extended with a bit more details and included a high level design to represent window operator state (along the same core principles as we had the keyed states)
Overall I feel a strong general consensus about the feature and key design principles. Given the scope and somewhat experimental nature of it, I expect to make some significant iterations over the coming months to converge to a good solution but we definitely have a really solid starting point. Unless there is further significant feedback, I will start the vote at the beginning of next week. Cheers Gyula On Fri, Jul 10, 2026 at 6:58 AM <[email protected]> wrote: > Hi Weiqing! > > Thanks for flagging this, I wasn’t aware of the other discussion details! > > First of all I agree that we have to decouple the storage of the names in > the snapshot from the schema evolution enabled/disabled logic . > > Given the obvious need for names by multiple flips and the straightforward > change , I think we can actually separate this into a jira and implement it > already independently. > > A small note on the schema evolution and maybe I need to review the design > and comment on the other FLIP, but it would be best to avoid baking config > logic into the snapshot itself and maybe providing access to the job config > during the state restore is a cleaner overall approach? > > Cheers > Gyula > > Sent from my iPhone > > > On 9 Jul 2026, at 22:23, Weiqing Yang <[email protected]> wrote: > > > > Hi Gyula, > > > > Thanks for putting this FLIP together. I'm very supportive of the goals > of > > FLIP-599. A robust State Catalog is a fantastic step forward for state > > observability and management. > > > > You and Shengkai landed on bumping RowDataSerializerSnapshot to include > > field names to support the catalog's schema inference. I agree with the > > direction, and FLIP-527 (State Schema Evolution for RowData) [1], which > I'm > > driving, proposes the exact same version bump. I wanted to flag a design > > overlap early, while both FLIPs are still in DISCUSS, so we can align our > > approaches. > > > > When implementing FLIP-527, one option is that the presence of these > field > > names acts as the explicit opt-in signal for schema evolution. The > feature > > is off by default, the serializer is built with names only when it's > > enabled, and the compatibility/migration logic keys off whether those > names > > are present. > > > > The catch with that approach is a coupling worth designing around. For > the > > catalog to recover real names on jobs that never enabled schema > evolution, > > those names would need to be written regardless of the evolution opt-in. > > But if name presence is what signals the opt-in, then writing names by > > default would flip the gate on for jobs that never asked, and trigger > > migration logic they didn't opt into. Since we're still discussing the > > high-level architecture for both FLIPs, this looks very solvable. We just > > need to decouple the "names persisted" metadata from the "evolution > > permitted" signal, perhaps by introducing an explicit toggle flag in the > > snapshot format rather than relying solely on name presence. That's the > > direction I'd favor for FLIP-527 as well. > > > > Getting this right while both FLIPs are in DISCUSS means whichever lands > > first can establish a format that serves both the catalog's metadata > needs > > and the migration's safety requirements. Looking forward to your thoughts > > on how best to converge these designs. > > > > Best, > > Weiqing > > > > [1] > > > https://cwiki.apache.org/confluence/spaces/FLINK/pages/353601981/FLIP-527+State+Schema+Evolution+for+RowData > > > > > >> On Wed, Jul 8, 2026 at 6:41 AM Jingsong Li <[email protected]> > wrote: > >> > >> Hi Gyula, > >> > >> Thanks for your response. > >> > >> Sounds good to me. > >> > >> Best, > >> Jingsong > >> > >>> On Wed, Jul 8, 2026 at 5:38 PM Gyula Fóra <[email protected]> > wrote: > >>> > >>> Hi Jingsong! > >>> > >>> Thanks for the feedback. > >>> I totally agree with you that the current design is going to be most > >> useful > >>> for datastream / user defined operators and may surface some "strange" > or > >>> slightly unexpected internals to the users for SQL jobs and SQL > >> operators. > >>> > >>> Big +1 for the idea that we should add operator specific views later to > >>> expose some important sql operator internals in a more human readable > >>> format. And I also completely agree that this is out of scope for the > >>> initial V1 version.Based on the interest I see on this FLIP I think > there > >>> will be a demand for a lot of similar extension capabilities once we > get > >>> the initial physical/raw state view version out. > >>> > >>> Regarding the documentation requirements for transparency, this is a > very > >>> important point and I couldn't agree more. In addition I think we will > >> have > >>> to clearly document and treat the contents of the state catalog as > fully > >>> experimental initially. With the great variety of internal > >> representations > >>> and use-cases surrounding Flink states, it's impossible to get this > right > >>> on the first try and our goal should be to iterate on this over a few > >> flink > >>> minor versions. > >>> > >>> So specifically: > >>> 1. The physical state representation of SQL and other built in > operators > >>> is a Flink internal and not part of the public api. Therefore the > >> structure > >>> / schema of these tables are subject to change without notice across > >> Flink > >>> versions. > >>> 2. The StateCatalog itself including configuration, operator/state to > >>> table mapping, metadata views, etc. are all initially experimental and > >>> provide no backward compatibility guarantees and can change without > >> notice > >>> initially. > >>> > >>> I think 1. will stay like this for the foreseeable future , we do not > >> want > >>> to make internal implementations as part of the public api. We should > aim > >>> to stabilize 2. and make it part of the public api eventually, and the > >>> views on the internal operators could hopefully become part of 2 as > well. > >>> > >>> Cheers > >>> Gyula > >>> > >>> On Wed, Jul 8, 2026 at 2:55 AM Jingsong Li <[email protected]> > >> wrote: > >>> > >>>> Hi Gyula and all, > >>>> > >>>> Thanks for driving this FLIP. I like the overall direction and think > >> making > >>>> savepoint/checkpoint state discoverable and queryable from SQL would > be > >>>> very > >>>> useful. > >>>> > >>>> I have one concern/question around how the catalog presents state from > >> SQL > >>>> operators. > >>>> > >>>> For many SQL operators, the internal state layout can be quite > >> different > >>>> from > >>>> the logical SQL model. For example, streaming joins, window > >> aggregations, > >>>> deduplication, TopN/rank, and other optimized operators may maintain > >>>> multiple > >>>> internal states, timers, namespaces, accumulators, or > >> optimization-specific > >>>> auxiliary structures. If these are exposed directly as SQL tables, the > >>>> result > >>>> may be difficult for users to understand. It may also accidentally > >> make an > >>>> internal state layout look like a stable user-facing contract, even > >> though > >>>> it > >>>> can change with planner/runtime optimizations or Flink versions. > >>>> > >>>> Would it make sense to explicitly distinguish two levels in the FLIP? > >>>> > >>>> 1. Physical/raw state views > >>>> > >>>> These are generic views derived from checkpoint/savepoint metadata and > >>>> serializer snapshots. They expose the actual operator UID, state name, > >>>> state > >>>> type, key/namespace/value schema, serializer information, etc. I think > >>>> this is > >>>> a very reasonable scope for the first version, especially for > >> debugging, > >>>> observability, migration, and advanced operational use cases. > >>>> > >>>> However, it would be good to document that these tables represent > >> Flink's > >>>> physical state layout and should not be treated as stable logical SQL > >>>> tables. > >>>> > >>>> 2. Optional logical/operator-aware views > >>>> > >>>> For some common SQL/runtime operators, we could later add > >> operator-specific > >>>> views or descriptors that explain the state in terms of operator > >> semantics. > >>>> For example, a join operator could expose left-side rows, right-side > >> rows, > >>>> and > >>>> timers in a more understandable way; a window aggregate could expose > >> window > >>>> accumulators and timer/namespace metadata. > >>>> > >>>> I do not think this needs to be part of the initial implementation, > but > >>>> making > >>>> the distinction explicit would help set the right expectations and > >> avoid > >>>> over-promising what automatic state-to-table mapping can provide. > >>>> > >>>> So my suggestion is that V1 focuses on the generic physical/raw state > >> view, > >>>> with clear metadata and documentation, while leaving > >> logical/operator-aware > >>>> views as a possible extension. > >>>> > >>>> Best, > >>>> Jingsong > >>>> > >>>> On Wed, Jul 8, 2026 at 3:39 AM Roman Khachatryan <[email protected]> > >> wrote: > >>>>> > >>>>>> 1. The catalog built on the regular state processor api (and > >> therefore > >>>>>> flink state restore) capabilities has limited scope to detect > >> exactly > >>>> what > >>>>>> happens when a state is no longer there. This will probably lead to > >>>> read > >>>>>> errors/not found exceptions etc, some of which happens in code > >> that is > >>>> a > >>>>>> bit tricky to control this way. Let's see how well this works in > >>>> practice > >>>>>> and error handling can always be improved in general. This is not > >> part > >>>> of > >>>>>> the catalog design itself. > >>>>> > >>>>> Makes sense. I think this can also be an extension. > >>>>> To clarify the ownership/pinning follow-up idea: it could use FS > >>>> mechanisms > >>>>> rather than HA (ZK/etcd). For example, a separate file > >>>>> with the lowest checkpoint that Flink should keep, protected by CAS > >> and > >>>>> limited to a specified TTL. > >>>>> > >>>>>> 2. The proposal in it's current form includes a global state > >> metadata > >>>> view > >>>>>> based on the existing metadata information ([1]) and based on > >>>> Shengkai's > >>>>>> feedback a per operator granular metadata table/view that would > >> expose > >>>>>> information of individual states. I don't see where file level > >>>> information > >>>>>> fits into this but if you have a good way / idea how to represent > >> this > >>>> as > >>>>> a > >>>>>> table this can definitely be a future extension/addition > >>>>> > >>>>> I was thinking about something like: > >>>>> > >>>>> USE `00000000000000000000/chk-42`; > >>>>> SELECT * > >>>>> FROM state_handles; > >>>>> > >>>>> -- id type parent path > >>>>> size timestamp operator/state/subtask_index > >>>>> local_path key_range > >>>>> -- 0 FileStateHandle - > >>>> s3://.../_metadata > >>>>> 0.8Kb 2026-07-02T22:56:30 - > >>>>> - - > >>>>> -- 1 IncrementalRemoteKeyedStateHandle 0 > >>>>> - 2026-07-02T22:56:14 > >>>>> uid_transaction_aggregator_keyed/users#0 - 0 .. 127 > >>>>> -- 2 FileStateHandle 1 > >>>>> s3://.../xxxx-xxxx... 5.4Mb 2026-07-02T22:56:12 - > >>>>> 000034.SST - > >>>>> -- 3 FileStateHandle 1 > >>>>> s3://.../yyyy-yyyy... 872Kb 2026-07-02T22:56:12 - > >>>>> 000035.SST - > >>>>> -- 4 IncrementalRemoteKeyedStateHandle 0 > >>>>> - 2026-07-02T22:56:24 > >>>>> uid_transaction_aggregator_keyed/users#1 - 128 .. > >> 255 > >>>>> -- 5 ByteStreamStateHandle 4 > >>>>> 2Kb 2026-07-02T22:56:24 - > >>>>> 000001.SST - > >>>>> > >>>>> The idea is to represent CompletedCheckpoint as a DAG so that it maps > >>>>> directly to the layout of the object in-memory. > >>>>> > >>>>>> 3. Did not think about this but if this becomes a requirement we > >> could > >>>> add > >>>>>> a flag to enable metadata only in the catalog. > >>>>> > >>>>> For our use-case (multi-tenant cloud environment), separate access > >> models > >>>>> for > >>>>> data and metadata are very likely a must have because of the security > >>>>> concerns: > >>>>> - internally, the operators should have access to metadata, but not > >> to > >>>> the > >>>>> customer data > >>>>> - externally, the users should have access to their data but not to > >> the > >>>>> metadata > >>>>> > >>>>> 5, 6. Thanks! :) > >>>>> > >>>>> 7. Yes, this should be available since metadata V4. > >>>>> > >>>>> Regards, > >>>>> Roman > >>>>> > >>>>> > >>>>> On Tue, Jul 7, 2026 at 9:56 AM Gyula Fóra <[email protected]> > >> wrote: > >>>>> > >>>>>> Hey Shengkai! > >>>>>> > >>>>>> Thanks for the questions, you hit on some very good practical > >> points. > >>>> Let > >>>>>> me provide my answers below, in the meantime I have already > >> updated the > >>>>>> FLIP to include some of your suggestions :) > >>>>>> > >>>>>> 1. How would schema inference work for RowDataSerializer? > >>>>>> > >>>>>> That's a good observation, I did not notice this. Probably the > >> simplest > >>>>>> solution would be to introduce a new version in the > >>>>>> RowDataSerializerSnapshot and include the names for this use-case. > >>>>>> This would not really impact checkpointing times/performance but > >> would > >>>>>> allow a straightforward mapping for sql states. > >>>>>> > >>>>>> If we feel that this is too much internal change, then we could > >> also > >>>> keep > >>>>>> it as is for now using simply f0, f1... > >>>>>> > >>>>>> 2. Is one keyed-state table per operator the right abstraction? > >>>>>> > >>>>>> This is a very good point and something that has bothered me as > >> well > >>>> from a > >>>>>> design perspective. There is no single good abstraction here > >> because > >>>> there > >>>>>> are completely different use cases. > >>>>>> When you just want to look into the state for a single / multiple > >> keys > >>>> and > >>>>>> you mostly have simple value list states, the single table > >>>> representation > >>>>>> is superior from both query syntax and performance perspective. It > >>>> avoids > >>>>>> JOINS and maps to the simple mental model that for a certain key > >> you > >>>> have > >>>>>> state x,y,z. Due to this straightforward mental model I still think > >>>> this is > >>>>>> the good default representation. With projection pushdown, it's > >> easy to > >>>>>> select one/several specific columns without reading / touching any > >>>> other. > >>>>>> > >>>>>> The big issue is however with large collection states that simply > >>>> cannot be > >>>>>> represented within a single row. This happens very often and is > >> one of > >>>> the > >>>>>> main reasons someone would even use a list state (if they > >> understand > >>>> how > >>>>>> they work internally, but not all users do...). Large map, window, > >> list > >>>>>> states won't work in the simple row model and are completely > >>>> impractical. > >>>>>> > >>>>>> Based on this, my recommendation would be to keep all keyed states > >> in a > >>>>>> single table as per the original proposal (one column per keyed > >> state) > >>>> but > >>>>>> also add an extra table per list / map state with the flattened > >> schema. > >>>>>> So if the operator has a value and list state, then there would be > >> 2 > >>>>>> tables. One with both states as columns (as per original design) + > >> 1 > >>>>>> flattened list state table (key, index, value) or for map states > >> (key, > >>>>>> map_key, value). > >>>>>> > >>>>>> This way we cover both use cases naturally. I am also open to > >> making > >>>> this > >>>>>> configurable on the catalog level. > >>>>>> > >>>>>> I have added this to the FLIP > >>>>>> > >>>>>> 3. Could you clarify the assumption of "reading state without user > >>>>>> classes"? > >>>>>> > >>>>>> Turns out from an implementation perspective it's not too bad and > >>>> pojo/avro > >>>>>> state schemas can be inferred quite naturally for most cases. > >> However > >>>> if > >>>>>> the user indeed provides the user jar on the classpath then the > >> whole > >>>>>> schema resolution will become even simpler because then we do not > >> need > >>>> any > >>>>>> custom inference. For our own use-cases and in general I would not > >>>> like to > >>>>>> assume that user classes will be easily available or that a catalog > >>>> will > >>>>>> represent mostly a single application. On the contrary the way We > >>>> intend to > >>>>>> use this, is definitely mostly without userjars and to represent > >>>> multiple > >>>>>> applications at the same time. > >>>>>> > >>>>>> 4. Could StateCatalog expose more fine-grained metadata? > >>>>>> > >>>>>> I think this is a very good idea. I have updated the FLIP to > >> include an > >>>>>> operator level metadata table as well (one for each operator). I > >> would > >>>> love > >>>>>> to include everything that you suggested, I think the practical > >> limit > >>>> is > >>>>>> what kind of information is part of the checkpoint and what isn't . > >>>> This > >>>>>> also ties to some questions Roman had about more detailed metadata. > >>>> Makes > >>>>>> sense > >>>>>> > >>>>>> Cheers > >>>>>> Gyula > >>>>>> > >>>>>> > >>>>>> On Tue, Jul 7, 2026 at 4:59 AM Shengkai Fang <[email protected]> > >>>> wrote: > >>>>>> > >>>>>>> Hi Gyula, > >>>>>>> > >>>>>>> Thanks for the FLIP. I like the direction of making > >>>> savepoint/checkpoint > >>>>>>> state discoverable and queryable from SQL. I have a few > >> questions and > >>>>>>> concerns about the proposed abstraction. > >>>>>>> > >>>>>>> 1. How would schema inference work for RowDataSerializer? > >>>>>>> > >>>>>>> From the current RowDataSerializer snapshot, it looks like the > >>>> snapshot > >>>>>>> persists the LogicalType[] and nested serializer snapshots, so > >> the > >>>> field > >>>>>>> types can be restored. However, the top-level RowDataSerializer > >>>>>> constructed > >>>>>>> from a RowType seems to store only the child LogicalTypes, not > >> the > >>>>>> RowType > >>>>>>> field names. Would StateCatalog expose generated names such as > >>>> f0/f1, or > >>>>>> is > >>>>>>> there another source for recovering the original field names? > >>>>>>> > >>>>>>> 2. Is one keyed-state table per operator the right abstraction? > >>>>>>> > >>>>>>> I wonder whether one table per named keyed state would be a > >> better > >>>> base > >>>>>>> abstraction, with an optional operator-level wide view on top. In > >>>>>>> particular, MapState can contain an unbounded or highly variable > >>>> number > >>>>>> of > >>>>>>> entries per state key. Exposing it as a MAP<K,V> column may > >> require > >>>> fully > >>>>>>> deserializing the map for a key into heap. A normalized table > >> shape > >>>> such > >>>>>>> as: > >>>>>>> > >>>>>>> (state_key, map_key, map_value) > >>>>>>> > >>>>>>> seems more scalable and SQL-friendly for MapState. Similarly, > >>>>>>> ValueState/ListState/MapState have different natural table > >> shapes, so > >>>>>> tying > >>>>>>> the physical table boundary to the operator may be too coarse. > >>>>>>> > >>>>>>> 3. Could you clarify the assumption of "reading state without > >> user > >>>>>>> classes"? > >>>>>>> > >>>>>>> This is a very attractive goal, but it also seems to introduce > >>>>>> substantial > >>>>>>> complexity for POJOs, Avro SpecificRecord, subclasses, and custom > >>>>>>> serializers. If StateCatalog is positioned as a > >>>>>> job-level/application-level > >>>>>>> catalog, would requiring the job jar or user artifacts be > >> acceptable > >>>> as a > >>>>>>> first step? That might simplify the design while still covering > >> many > >>>>>>> operational/debugging use cases. > >>>>>>> > >>>>>>> 4. Could StateCatalog expose more fine-grained metadata? > >>>>>>> > >>>>>>> For debugging state, it would be useful to expose state-level > >>>> metadata > >>>>>> such > >>>>>>> as state name, state type, serializer snapshot/serializer class, > >> TTL > >>>>>>> configuration, namespace/window information where applicable, > >> backend > >>>>>> state > >>>>>>> type, and possibly whether a state can be read > >> lazily/streamingly. > >>>>>>> > >>>>>>> Best, > >>>>>>> Shengkai > >>>>>>> > >>>>>>> Roman Khachatryan <[email protected]> 于2026年7月7日周二 08:44写道： > >>>>>>> > >>>>>>>> Hi Gyula, > >>>>>>>> > >>>>>>>> Thanks for the proposal, this looks very useful! A few > >> questions > >>>> and > >>>>>>>> comments: > >>>>>>>> > >>>>>>>> 1. Following up on Han's question about checkpoint retention: I > >>>>>>> understand > >>>>>>>> external coordination is out of scope for now, but could the > >>>> catalog at > >>>>>>>> least detect that a checkpoint was subsumed/deleted mid-query > >> and > >>>> fail > >>>>>>> with > >>>>>>>> a clear error, rather than a low-level file-not-found? And do > >> you > >>>> see > >>>>>>>> ownership/pinning as a possible follow-up FLIP once checkpoint > >>>> reading > >>>>>>>> picks up adoption? > >>>>>>>> 2. Does the proposal allow querying file-level metadata (file > >> size, > >>>>>>>> creation date, etc.)? This would be useful for debugging > >>>>>>> compaction-related > >>>>>>>> issues. > >>>>>>>> 3. If yes, could data and metadata queries have separate access > >>>> modes? > >>>>>> In > >>>>>>>> many environments access to data is much stricter than access > >> to > >>>>>>> metadata, > >>>>>>>> so being able to grant metadata-only access to the catalog > >> would > >>>>>> broaden > >>>>>>>> where it can be deployed. > >>>>>>>> 4. Just to confirm: incremental checkpoints are expected to > >> work > >>>>>> through > >>>>>>>> the regular restore mechanisms, given sufficient retention? > >>>>>>>> 5. +1 on bringing non-keyed state into scope — a concrete use > >> case: > >>>>>>>> inspecting Kafka transaction state (currently stored in > >> non-keyed > >>>>>>> operator > >>>>>>>> state) would be very valuable for debugging EOS issues. > >>>>>>>> 6. Could you explain why timers are not supported? They live in > >>>> keyed > >>>>>>> state > >>>>>>>> and the state processor API can read registered timers, so I'm > >>>>>> wondering > >>>>>>>> whether this is a fundamental limitation or just table-mapping > >>>> scope. > >>>>>>>> 7. Does the proposal allow querying checkpoint metadata (such > >> as > >>>>>>>> SharingFilesStrategy, isSavepoint, etc.)? This could be useful > >> for > >>>>>>>> debugging CLAIM mode issues. > >>>>>>>> > >>>>>>>> > >>>>>>>> Regards, > >>>>>>>> Roman > >>>>>>>> > >>>>>>>> > >>>>>>>> On Mon, Jul 6, 2026 at 1:02 PM Gyula Fóra < > >> [email protected]> > >>>>>> wrote: > >>>>>>>> > >>>>>>>>> Hi Zakelly! > >>>>>>>>> > >>>>>>>>> That's a good point and we have to ensure that it works. In > >>>> theory > >>>>>> SQL > >>>>>>>>> related states are relatively easy to cover and represent. > >> The > >>>>>> RowData > >>>>>>>>> state would be mapped directly to ROW<...> similar to other > >> pojo > >>>> key > >>>>>>>>> states. > >>>>>>>>> > >>>>>>>>> Cheers > >>>>>>>>> Gyula > >>>>>>>>> > >>>>>>>>> On Sun, Jul 5, 2026 at 4:03 PM Zakelly Lan < > >>>> [email protected]> > >>>>>>>> wrote: > >>>>>>>>> > >>>>>>>>>> Hi Gyula, > >>>>>>>>>> > >>>>>>>>>> Thanks for driving this, it's a nice addition and I fully > >>>> support > >>>>>> it. > >>>>>>>> One > >>>>>>>>>> thing to make sure: > >>>>>>>>>> > >>>>>>>>>> For the state generated by some Flink SQL jobs, does the > >>>>>> StateCatalog > >>>>>>>>> infer > >>>>>>>>>> this internal `RowData` structure and expose it as a SQL > >>>> `ROW<...>` > >>>>>>>> type? > >>>>>>>>>> For example, a regular streaming join side may be stored > >> as a > >>>> state > >>>>>>>> such > >>>>>>>>> as > >>>>>>>>>> `left-records` / `right-records`, whose value or map > >> key/value > >>>>>>>> contains a > >>>>>>>>>> `RowData` for the original input row. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> Best, > >>>>>>>>>> Zakelly > >>>>>>>>>> > >>>>>>>>>> On Fri, Jul 3, 2026 at 4:13 PM Dennis-Mircea Ciupitu < > >>>>>>>>>> [email protected]> wrote: > >>>>>>>>>> > >>>>>>>>>>> Hi Gyula, > >>>>>>>>>>> > >>>>>>>>>>> Thanks for the detailed answers. This addresses my > >> questions > >>>> well > >>>>>>> and > >>>>>>>>> the > >>>>>>>>>>> direction sounds great. > >>>>>>>>>>> > >>>>>>>>>>> +1 (non-binding) from my side. > >>>>>>>>>>> > >>>>>>>>>>> Best regards, > >>>>>>>>>>> Dennis > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> On Thu, Jul 2, 2026 at 3:26 PM Gyula Fóra < > >>>> [email protected]> > >>>>>>>>> wrote: > >>>>>>>>>>> > >>>>>>>>>>>> Hi Dennis! > >>>>>>>>>>>> > >>>>>>>>>>>> Thank you for the questions. Much recent work in the > >> state > >>>>>>>> connector > >>>>>>>>>> api > >>>>>>>>>>>> has been done basically towards this type of nice > >>>> cataloging > >>>>>> and > >>>>>>>>>> flexible > >>>>>>>>>>>> access. There are a few holes and things that have to > >> be > >>>>>> changed, > >>>>>>>> not > >>>>>>>>>>>> everything is enumerated in the FLIP but we have to > >> have an > >>>>>> open > >>>>>>>> mind > >>>>>>>>>> and > >>>>>>>>>>>> make all necessary changes as you said to make this > >> truly > >>>> nice > >>>>>>> and > >>>>>>>>>>>> comprehensive as much as possible. Most state processor > >>>> apis > >>>>>> are > >>>>>>>>> marked > >>>>>>>>>>>> experimental so we can be flexible within reason :) > >>>>>>>>>>>> > >>>>>>>>>>>> Now to the concrete questions: > >>>>>>>>>>>> > >>>>>>>>>>>> 1. Non-keyed state support / scope > >>>>>>>>>>>> I think non-keyed states should definitely be in the > >> scope > >>>> of > >>>>>> the > >>>>>>>>> FLIP > >>>>>>>>>> in > >>>>>>>>>>>> terms of design , and my intention was not to exclude > >> them > >>>> I > >>>>>> just > >>>>>>>>>> focused > >>>>>>>>>>>> on the keyed state as that is readily available in our > >>>>>> prototype > >>>>>>>>>>>> implementation (without much changes to the existing > >>>>>>> connectors). I > >>>>>>>>>> will > >>>>>>>>>>>> try to update the FLIP to include non-keyed states > >> more in > >>>>>> detail > >>>>>>>>> but I > >>>>>>>>>>>> think the case is pretty straightforward. From a table > >>>>>>>> representation > >>>>>>>>>>>> perspective, they can follow a similar pattern such as: > >>>>>>>>>>>> uid_opUID_statename_broadcast , > >> uid_opUID_statename_list > >>>> . A > >>>>>>>>>>> corresponding > >>>>>>>>>>>> SQL connector can easily be added to support these > >> based > >>>> on the > >>>>>>>>>> existing > >>>>>>>>>>>> datastream connector. I will make sure to add separate > >>>> tickets > >>>>>>> for > >>>>>>>>>> these > >>>>>>>>>>>> types of states once the FLIP is accepted and this > >> work can > >>>>>> very > >>>>>>>>> easily > >>>>>>>>>>> be > >>>>>>>>>>>> parallelized across different state types within the > >>>> existing > >>>>>>>> catalog > >>>>>>>>>>>> frameworks. This way keyed/non-keyed states will live > >>>> directly > >>>>>>>>> together > >>>>>>>>>>> in > >>>>>>>>>>>> a single catalog/db. > >>>>>>>>>>>> > >>>>>>>>>>>> In the future we can even go a step further and include > >>>>>> connector > >>>>>>>>>>> specific > >>>>>>>>>>>> state views such as kafka offsets etc with custom > >> connector > >>>>>>>> specific > >>>>>>>>>>>> plugins > >>>>>>>>>>>> > >>>>>>>>>>>> 2/3. Serializer transparency and robustness > >>>>>>>>>>>> From a practical standpoint both generated (synthetic) > >>>>>>> serializers > >>>>>>>>> and > >>>>>>>>>>>> custom classes / kryo and pluggable logic could work > >> but > >>>> the > >>>>>>> whole > >>>>>>>>>>> catalog > >>>>>>>>>>>> concepts requires a certain behaviour to be useful. The > >>>> catalog > >>>>>>>> would > >>>>>>>>>>> point > >>>>>>>>>>>> to savepoint directories and discover all state in it > >>>>>>> (potentially > >>>>>>>>> from > >>>>>>>>>>>> multiple jobs). Configuration has to be done in a > >> generic > >>>> way, > >>>>>> I > >>>>>>>>> don't > >>>>>>>>>>> see > >>>>>>>>>>>> a problem with introducing configs for specifying > >> custom > >>>>>>>>>>>> serializers/factories either generically for certain > >>>> specific > >>>>>>>>> classes. > >>>>>>>>>> In > >>>>>>>>>>>> most cases however this won't be necessary as the state > >>>>>> snapshot > >>>>>>>>> itself > >>>>>>>>>>>> usually has a reference (classname) of the original > >> user > >>>>>> classes. > >>>>>>>> If > >>>>>>>>>> the > >>>>>>>>>>>> catalog process has access to those classes it will use > >>>> that > >>>>>>>>> directly, > >>>>>>>>>> or > >>>>>>>>>>>> other confugred serializers, and only if not available > >> fall > >>>>>> back > >>>>>>> to > >>>>>>>>>>>> generating serializers for POJO/TUPLE types. There is > >>>>>> obviously a > >>>>>>>>> limit > >>>>>>>>>>> to > >>>>>>>>>>>> what is possible here initially, Kryo being one > >> exception > >>>> where > >>>>>>> you > >>>>>>>>>>> either > >>>>>>>>>>>> have the class or not. > >>>>>>>>>>>> > >>>>>>>>>>>> I would like to however point out that we do not have > >> to > >>>>>> support > >>>>>>>>>>> everything > >>>>>>>>>>>> initially, we can start with what is currently > >> available, > >>>> use > >>>>>> the > >>>>>>>>>>> classpath > >>>>>>>>>>>> / generated serializers and as we develop we will find > >> the > >>>>>> limits > >>>>>>>> of > >>>>>>>>>> this > >>>>>>>>>>>> approach and then can extend with configuration as it > >> feels > >>>>>>> natural > >>>>>>>>>>> instead > >>>>>>>>>>>> of trying to create a super complex initial solution. > >> But I > >>>>>>>>> definitely > >>>>>>>>>>>> agree that we should support custom serializer already > >>>>>> specified > >>>>>>> in > >>>>>>>>> the > >>>>>>>>>>>> config that is otherwise used by flink for the jobs > >> (but I > >>>>>> think > >>>>>>>> this > >>>>>>>>>>>> should more or less work out of the box). > >>>>>>>>>>>> > >>>>>>>>>>>> 4. The metadata view is currently reused based on the > >>>> existing > >>>>>>>> table > >>>>>>>>>>> valued > >>>>>>>>>>>> function. Let's take this as a followup under this > >>>> umbrella to > >>>>>>>>> improve > >>>>>>>>>> / > >>>>>>>>>>>> extend the metadata view. I don't think we need a > >> separate > >>>> FLIP > >>>>>>> but > >>>>>>>>> it > >>>>>>>>>>> also > >>>>>>>>>>>> feels out of scope here. > >>>>>>>>>>>> > >>>>>>>>>>>> Cheers > >>>>>>>>>>>> Gyula > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> On Thu, Jul 2, 2026 at 1:02 PM Dennis-Mircea Ciupitu < > >>>>>>>>>>>> [email protected]> wrote: > >>>>>>>>>>>> > >>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Thank you for driving this. Being able to discover > >>>>>>>>>>> savepoints/checkpoints > >>>>>>>>>>>>> and query their state as SQL tables without shipping > >> the > >>>>>>> original > >>>>>>>>>> user > >>>>>>>>>>>>> classes is a genuinely valuable addition, and it's > >> nice > >>>> that > >>>>>> it > >>>>>>>>>> builds > >>>>>>>>>>> on > >>>>>>>>>>>>> the existing state-table connector and > >> savepoint_metadata > >>>>>> work > >>>>>>>>> rather > >>>>>>>>>>>> than > >>>>>>>>>>>>> starting from scratch. > >>>>>>>>>>>>> > >>>>>>>>>>>>> A few points and questions, mostly around scope and > >> the > >>>>>>>> serializer > >>>>>>>>>>> story: > >>>>>>>>>>>>> > >>>>>>>>>>>>> 1. Non-keyed state and the DataStream path. > >>>>>>>>>>>>> - The FLIP scopes out BroadcastState, operator > >>>>>> ListState > >>>>>>>> and > >>>>>>>>>>>>> UnionState because "no readily available Table > >> API > >>>>>>>> connectors > >>>>>>>>>>> exist > >>>>>>>>>>>>> for > >>>>>>>>>>>>> these state types." That's a fair > >> characterization > >>>> of > >>>>>> the > >>>>>>>>> Table > >>>>>>>>>>>>> layer, but > >>>>>>>>>>>>> the state-processor DataStream API already > >> reads > >>>> all > >>>>>>> three > >>>>>>>>>> today > >>>>>>>>>>>>> (SavepointReader#readBroadcastState / > >>>> #readUnionState / > >>>>>>>>>>>>> #readListState). So > >>>>>>>>>>>>> the limitation is really in the keyed-only SQL > >>>> mapping > >>>>>>>>>>>>> (KeyedStateReader > >>>>>>>>>>>>> runs inside a keyed backend), not in the > >> snapshots > >>>>>>>>> themselves. > >>>>>>>>>>>>> - Is the keyed-only scope a deliberate > >>>> UX/table-mapping > >>>>>>>>>> decision, > >>>>>>>>>>>> or > >>>>>>>>>>>>> would a DataStream-backed reader be considered > >> so > >>>> the > >>>>>>>> catalog > >>>>>>>>>>> isn't > >>>>>>>>>>>>> strictly less capable than the API it extends? > >>>> Even if > >>>>>>>>>> non-keyed > >>>>>>>>>>>>> contents > >>>>>>>>>>>>> stay out of scope initially, it would be good > >> to > >>>> frame > >>>>>>> this > >>>>>>>>>>>>> explicitly as a > >>>>>>>>>>>>> Table-mapping constraint rather than a general > >> one. > >>>>>>>>>>>>> 2. Serializer transparency - the "no user classes" > >>>> premise > >>>>>>> vs. > >>>>>>>>>>> custom > >>>>>>>>>>>>> serializers. > >>>>>>>>>>>>> - The design relies on Flink's transparent > >>>> serializer > >>>>>>>> formats > >>>>>>>>>> to > >>>>>>>>>>>>> decode state without user dependencies, which > >> is > >>>> great > >>>>>>> for > >>>>>>>>>>>>> POJO/Avro/basic > >>>>>>>>>>>>> types. But two serialization efforts point the > >>>> other > >>>>>> way: > >>>>>>>>>>> FLIP-398 > >>>>>>>>>>>>> [1] > >>>>>>>>>>>>> (released) already lets users configure > >>>> serializers per > >>>>>>>> type > >>>>>>>>>> via > >>>>>>>>>>>>> pipeline.serialization-config, and FLIP-538 > >> [2] (in > >>>>>>>>> discussion) > >>>>>>>>>>>> adds > >>>>>>>>>>>>> pluggable custom generic-type serializers (e.g. > >>>> Apache > >>>>>>>> Fory) > >>>>>>>>>> and > >>>>>>>>>>>>> promotes > >>>>>>>>>>>>> TypeSerializer/TypeSerializerSnapshot to > >> @Public. > >>>> As > >>>>>>>> FLIP-538 > >>>>>>>>>>>>> itself notes, > >>>>>>>>>>>>> state written with a custom serializer becomes > >>>>>> dependent > >>>>>>> on > >>>>>>>>>> that > >>>>>>>>>>>>> serializer > >>>>>>>>>>>>> to decode - external tooling without it cannot > >> read > >>>>>> those > >>>>>>>>>> bytes. > >>>>>>>>>>>>> - Could we make the deserialization side > >> pluggable > >>>> and > >>>>>>>>>>>> config-driven, > >>>>>>>>>>>>> mirroring FLIP-398's serialization-config, > >> with a > >>>>>>> graceful > >>>>>>>>>>> fallback > >>>>>>>>>>>>> (e.g. > >>>>>>>>>>>>> expose the raw bytes / skip the column) when a > >>>> format > >>>>>>> isn't > >>>>>>>>>>>>> transparently > >>>>>>>>>>>>> decodable? There already seems to be a seam for > >>>> this > >>>>>>>>>>>>> (SavepointTypeInformationFactory), and making > >> it a > >>>>>>>>> first-class, > >>>>>>>>>>>>> config-selectable option would keep the catalog > >>>>>>>>>>> forward-compatible > >>>>>>>>>>>> as > >>>>>>>>>>>>> serialization support grows. > >>>>>>>>>>>>> 3. Robustness of the transparent decoding path. > >>>>>>>>>>>>> - Related to (2): reconstructing values by > >>>> mirroring > >>>>>> the > >>>>>>>>> binary > >>>>>>>>>>>>> layout (PojoToRowDataDeserializer) is the most > >>>> powerful > >>>>>>> but > >>>>>>>>>> also > >>>>>>>>>>>> the > >>>>>>>>>>>>> most > >>>>>>>>>>>>> fragile part of the design. How is it expected > >> to > >>>>>> behave > >>>>>>>>> across > >>>>>>>>>>>>> serializer > >>>>>>>>>>>>> schema evolution / state migration (a > >> serializer > >>>>>> snapshot > >>>>>>>>> that > >>>>>>>>>>>>> differs from > >>>>>>>>>>>>> the writer's), Kryo-fallback fields, > >> nested/generic > >>>>>>> types, > >>>>>>>>> and > >>>>>>>>>>>>> nullability? > >>>>>>>>>>>>> - It would help to spell out the supported > >> matrix > >>>> and > >>>>>> the > >>>>>>>>>> failure > >>>>>>>>>>>>> mode (hard error vs. degrade to raw bytes) up > >>>> front, > >>>>>>> since > >>>>>>>>> this > >>>>>>>>>>>>> is exactly > >>>>>>>>>>>>> where "read without the user classes" is most > >>>> likely to > >>>>>>>> break > >>>>>>>>>> in > >>>>>>>>>>>>> practice. > >>>>>>>>>>>>> 4. Observability / summary reporting. > >>>>>>>>>>>>> - The metadata view is a great start. Two small > >>>> asks: > >>>>>>>>>>>>> - per-subtask (or per-key-group) size > >>>> granularity in > >>>>>>>>>> addition > >>>>>>>>>>> to > >>>>>>>>>>>>> per-operator, since skew is usually what > >> you are > >>>>>>> chasing > >>>>>>>>> on > >>>>>>>>>>>>> large state; > >>>>>>>>>>>>> - optionally rounding out the size breakdown > >>>> with > >>>>>>>>>> managed/raw > >>>>>>>>>>>>> operator state and channel state sizes for a > >>>> full > >>>>>>>> picture > >>>>>>>>>>>> (noting > >>>>>>>>>>>>> the > >>>>>>>>>>>>> latter are in-flight / unaligned-checkpoint > >>>> buffers > >>>>>>>> rather > >>>>>>>>>>>>> than user state). > >>>>>>>>>>>>> - A prominent upfront summary of the largest > >>>> operators > >>>>>> / > >>>>>>>>> state > >>>>>>>>>> is > >>>>>>>>>>>>> often what users want before drilling in. > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> Best regards, > >>>>>>>>>>>>> Dennis > >>>>>>>>>>>>> > >>>>>>>>>>>>> [1] > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>> > >> > https://cwiki.apache.org/confluence/spaces/FLINK/pages/282102217/FLIP-398+Improve+Serialization+Configuration+And+Usage+In+Flink > >>>>>>>>>>>>> [2] > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>> > >> > https://cwiki.apache.org/confluence/spaces/FLINK/pages/373886828/FLIP-538+Support+Custom+Generic+Type+Serializer > >>>>>>>>>>>>> > >>>>>>>>>>>>> On Mon, Jun 29, 2026 at 12:53 PM Gyula Fóra < > >>>>>> [email protected] > >>>>>>>> > >>>>>>>>>> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>>> Hi Flink Devs! > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> I would like to start the discussion about > >> FLIP-599: > >>>> State > >>>>>>>>> Catalog > >>>>>>>>>>> [1] > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> State and stateful processing has always been one > >> of > >>>> the > >>>>>> most > >>>>>>>>>>>> fundamental > >>>>>>>>>>>>>> features of Flink and a major contributor to its > >>>> success > >>>>>> and > >>>>>>>>> global > >>>>>>>>>>>>>> adoption. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Over the years several apis and methods have been > >>>> developed > >>>>>>> to > >>>>>>>>>>> address > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> need for external access and analytics such as the > >>>> state > >>>>>>>>> processor > >>>>>>>>>>>>>> datastream / java apis, the since deprecated > >> queryable > >>>>>> state > >>>>>>>>>>>> abstractions > >>>>>>>>>>>>>> and more recently a number of table / SQL api > >>>> connectors to > >>>>>>>>> access > >>>>>>>>>>>> state > >>>>>>>>>>>>>> metadata and keyed states in a somewhat limited > >> way. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Extending the current capabilities of the > >>>>>> state-process-api, > >>>>>>>> this > >>>>>>>>>>> FLIP > >>>>>>>>>>>>> aims > >>>>>>>>>>>>>> to lift state processing, analytics and > >> observability > >>>> to a > >>>>>>> new > >>>>>>>>>> level > >>>>>>>>>>>> by > >>>>>>>>>>>>>> introducing the State Catalog. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> State Catalog is a Flink SQL Catalog implementation > >>>> that > >>>>>>> allows > >>>>>>>>>>>>> discovering > >>>>>>>>>>>>>> savepoints/checkpoints and mapping their state > >>>>>> automatically > >>>>>>> to > >>>>>>>>> SQL > >>>>>>>>>>>>> tables. > >>>>>>>>>>>>>> The tables are derived for the different operators > >> and > >>>>>> their > >>>>>>>>> keyed > >>>>>>>>>>>> states > >>>>>>>>>>>>>> with schema matching the state structure. Most > >>>> importantly > >>>>>> it > >>>>>>>>>>> supports > >>>>>>>>>>>>>> reading POJO / Avro and other structured and basic > >> type > >>>>>>> states > >>>>>>>>>>> without > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> original user classes (dependencies) by relying on > >>>> Flink's > >>>>>>>>>>> transparent > >>>>>>>>>>>>> and > >>>>>>>>>>>>>> efficiently structured serializer formats. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> We have a fully functional prototype implementation > >>>>>> developed > >>>>>>>>> with > >>>>>>>>>>>> Gabor > >>>>>>>>>>>>>> Somogyi that we will be happy to share if the > >> community > >>>>>>> accepts > >>>>>>>>> the > >>>>>>>>>>>>>> proposal! > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Looking forward to your feedback and suggestions! > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Gyula > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> [1] > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>> > >> > https://cwiki.apache.org/confluence/spaces/FLINK/pages/438009922/FLIP-599+State+Catalog > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>> > >> >
