gyfora opened a new pull request, #28837: URL: https://github.com/apache/flink/pull/28837
# State Catalog Initial Contribution This PR represents the initial code contribution for the State Catalog work as described in https://cwiki.apache.org/confluence/spaces/FLINK/pages/438009922/FLIP-599+State+Catalog This is already a massive PR but I felt that it is important to have at least the minimal functional pieces in place to provide enough context for reviewers instead of merging a bunch of utilities/changes that are nowhere used in the codebase. So for this initial code dump, the scope is the following: 1. Introduce the StateCatalog itself including savepoint/checkpoint discovery 2. Add required serializer / snapshot changes for state schema inference and reading without original user classes 3. Introduce KeyedState type inference / reading / cataloging I deliberately structured the code/changes in the below commits in a way that commits are probably easiest to review in order, independently instead of the whole PR together. Along the way several improvements and changes have been made to the existing Savepoint SQL connector, dropping / simplifying some connector options etc as we have discussed with @gaborgsomogyi offline. ## What is the purpose of the change This PR adds keyed-state table support to the `StateCatalog` (introduced in FLINK-40176), letting keyed operator state from a checkpoint/savepoint be read as SQL rows without requiring the original POJO/Avro classes on the classpath. In this initial version two table shapes are supported: - **General keyed-state table** (`STATE_READER_MODE.KEYED`): one row per key, with one column per registered keyed state (`ValueState` as a scalar column, `ListState`/`MapState` as `ARRAY`/`MAP` columns). - **Flattened keyed-state table** (`STATE_READER_MODE.KEYED_FLAT`): a single named `ListState`/`MapState` exposed as `(key, list_index, list_value)` or `(key, map_key, map_value)` rows. Schema discovery works by reading the raw keyed-state serialization header (via `StateSchemaExtractor`) and mapping `TypeSerializerSnapshot` trees to Flink SQL `LogicalType` (via `SerializerSnapshotToLogicalTypeConverter`), so tables can be inferred and read even when the POJO/Avro classes used at write time aren't available at read time. ## Brief change log Each standalone commit contains a brief description of the changes. ## Verifying this change This change added tests and can be verified as follows: - `StateTableUtilsTest`, `StateSchemaExtractorTest`, `SerializerSnapshotToLogicalTypeConverterTest`: unit coverage for schema discovery and type inference. - `PojoToRowDataDeserializerTest`, `InternalTypeConverterTest`, `PojoSerializerSnapshotLenientReadTest`: unit coverage for the POJO-to-RowData deserialization path. - `SavepointTypeInfoResolverTest`, `TypeConversionDriftGuardTest`, `SavepointDynamicTableSourceTest`: unit coverage for table mapping/type resolution and the DynamicTableSource. - `PojoStateSchemaExtractionITCase` (+ HashMap/RocksDB variants), `StateCatalogSqlITCase` (+ HashMap/RocksDB variants): integration tests reading real generated savepoints through SQL. - `StateCatalogITCase`: end-to-end integration tests exercising the general and flattened keyed-state tables via both direct `StateTableUtils`/`CatalogTable` construction and `StateCatalog` + SQL, across primitive, POJO, Avro, and Tuple key/value shapes, with schema-discovery-only (missing classes) and projection/filter push-down scenarios. In addition to the above unit/integration test coverage the core logic has been validated on kubernetes environments on different jobs and states including both memory and RocksDB state backend. Needless to say that major new features and additions like this always carry bugs that we have to fix over time as we improve our test coverage and the understanding of how this will be used in practice. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): yes — `flink-state-processing-api` now has a compile-scope, optional dependency on `flink-avro`. - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: yes — `PojoSerializerSnapshot`, `AvroSerializerSnapshot`, and `RowDataSerializerSnapshot` gain accessors used for schema conversion; no wire-format or compatibility changes. - The runtime per-record code paths (performance sensitive): no — changes are confined to the state-processor-api's offline reading path. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes — keyed-state SQL tables (general and flattened) for the state processor API's `StateCatalog`. - If yes, how is the feature documented? not documented yet ##### Was generative AI tooling used to co-author this PR? - [ x ] Yes - Claude Sonnet 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]
