[FLINK-40175][State Catalog] State Catalog initial catalog functionality and keyed state support - #28837
Open
gyfora wants to merge 6 commits into
Open
[FLINK-40175][State Catalog] State Catalog initial catalog functionality and keyed state support#28837gyfora wants to merge 6 commits into
gyfora wants to merge 6 commits into
Conversation
…ionality Introduces StateCatalog, a read-only Flink SQL catalog that discovers checkpoints/savepoints from configured directories and exposes their metadata as queryable databases and views. This first slice covers catalog registration (StateCatalogFactory/StateCatalogOptions), directory scanning and database naming (SnapshotDiscovery), and the per-snapshot "metadata" view backed by the savepoint_metadata table function. Keyed/non-keyed state table support is added in later commits.
…ce for keyed state Introduces the core utilities used to build keyed state catalog tables: StateSchemaExtractor reads the raw keyed-state serialization header to determine registered states without loading user POJO classes, SerializerSnapshotToLogicalTypeConverter maps TypeSerializerSnapshot trees to Flink SQL LogicalType, and StateTableUtils ties these together to classify an operator keyed states and build the corresponding CatalogTable, including best-effort state backend detection. Windowed, flattened, and non-keyed state support are intentionally left out and will be added in later commits.
… type inference Adds the serializer-level support needed to convert savepoint state into Flink table rows without the original classes on the classpath: - PojoSerializerSnapshot exposes its field serializer snapshots and registered subclass snapshots, and PojoDeserializerCompatibilitySnapshot lets a PojoToRowDataDeserializer round-trip through snapshotConfiguration()/restoreSerializer(). - AvroSerializerSnapshot exposes its Avro schema for conversion to a LogicalType. - RowDataSerializerSnapshot exposes its stored types and field names (getTypes()/getFieldNames()) for conversion to a LogicalType. - PojoToRowDataDeserializer reads the POJO binary format produced by PojoSerializer and produces GenericRowData directly, using InternalTypeConverter to convert individual field values to their table-internal representation. flink-state-processing-api now depends on flink-avro as a compile-scope (optional) dependency, since SerializerSnapshotToLogicalTypeConverter references Avro serializer snapshot classes directly.
… and provider interfaces Introduces the general (non-flattened) keyed-state table: each row exposes a key column plus one column per registered keyed state (ValueState as a scalar column, ListState/MapState as ARRAY/MAP columns). - SavepointConnectorOptions gains StateReaderMode (KEYED plus placeholders for the windowed/flattened/non-keyed modes added in later commits) and FLATTENED_STATE_NAME, replacing the legacy STATE_TYPE/KEY_CLASS/VALUE_CLASS options. - StateTableMapping resolves a table's schema-declared columns against the savepoint's actual registered states and their serializers, backed by shared helpers in KeyedTableMappingSupport and SavepointTypeInfoResolver (renamed/reworked from SavepointTypeInformationFactory). - StateValueConverter converts a runtime state value into its table-internal representation according to a resolved StateValueColumnConfiguration. - AbstractSavepointDynamicTableSource/AbstractMultiColumnScanProvider/ AbstractSavepointDataStreamScanProvider provide the shared scan-provider and DynamicTableSource machinery (projection/filter push-down, DataStream program construction) parameterized over a table mapping; MultiColumnStateMapping is the shared interface for "general", multi-value-column table mappings. SavepointDynamicTableSourceFactory wires STATE_READER_MODE.KEYED to this machinery via StateTableMapping. - KeyedStateReader reads the resolved states via the DataStream state processor API and emits rows accordingly. - StateReaderOperator/KeyedStateReaderOperator/MultiStateKeyIterator are extended to also expose the key-group of each returned key (needed to correctly read from the keyed backend without a keyed context), via a new getKeysAndKeyGroups method plumbed through KeyedStateBackend and its heap/ RocksDB/changelog/ForSt/batch-execution implementations. KeyedStateInputFormat additionally wires a POJO-to-RowData restore path via PojoSerializerSnapshot#setPojoRestoreSerializerFactory, and shares ExecutionConfig deserialization via the new ExecutionConfigs helper. WindowReaderOperator is mechanically adapted to the resulting StateReaderOperator signature change; it is unrelated to the new window table support added in later commits. - SavepointLoader gains loadOperatorMetadata, loading both the per-state serializer snapshots and the backend key serializer snapshot for an operator in a single I/O pass.
…g and implementation Adds the flattened keyed-state table (selected via STATE_READER_MODE.KEYED_FLAT): a single named LIST/MAP state is exposed as (key, list_index, list_value) or (key, map_key, map_value) rows instead of one column per state. - FlattenedStateTableMapping resolves and validates the fixed 3-column schema against the target state's actual serializers. - SingleColumnStateMapping is the shared interface for "flattened", single value-column table mappings (implemented by FlattenedStateTableMapping now, and reused by the windowed-flattened mapping added in a later commit). - AbstractSingleColumnScanProvider/FlattenedSavepointDynamicTableSource provide the shared scan-provider and DynamicTableSource machinery parameterized over a SingleColumnStateMapping, mirroring AbstractMultiColumnScanProvider/ AbstractSavepointDynamicTableSource from the previous commit. - FlattenedKeyedStateReader reads the target state via the DataStream state processor API and flattens each key's list/map entries into rows. - SavepointDynamicTableSourceFactory wires STATE_READER_MODE.KEYED_FLAT to this machinery, and KeyedTableMappingSupport gains the flattened-schema inference/validation and serializer-resolution helpers shared with the windowed-flattened mapping.
Collaborator
Adds end-to-end integration tests for the general and flattened keyed-state tables introduced in the previous two commits, exercising real (generated) savepoint fixtures end-to-end via both direct StateTableUtils/CatalogTable construction and StateCatalog + SQL: - testReadKeyedStateFromSchemaDiscovery / testReadAvroKeyedStateFromSchemaDiscovery / testSchemaExtractionWithoutPojoClass: schema discovery and data reading for a savepoint whose POJO/Avro classes are unavailable on the classpath. - testKeyedStateCatalog / testPojoAndAvroKeyedStateTables / testTupleKeyedStateTables: full StateCatalog-backed catalog/table discovery and SQL reads across primitive, POJO, Avro-specific, Avro-generic, and Tuple key/value shapes. - testFlattenedKeyedStateTables: flattened LIST/MAP state table schema and SQL reads, including state_key filter push-down. - testProjectionColumnReorder / testProjectionSubsetKeyOnly / testProjectionSubsetValueOnly: projection push-down correctness for the general keyed-state table. - testPojoAndAvroKeySchemaTypes / testTupleKeySchemaTypes: key-type schema inference for POJO, Avro-specific, and Tuple key types. The savepoint fixtures are produced by the generator programs under src/test/resources/generator (not run as part of the build; their output is checked in), plus the pre-existing missing-class/missing-avro fixtures from the schema-discovery bootstrapping commit.
gyfora
force-pushed
the
state-catalog-contrib-1
branch
from
July 29, 2026 12:09
18e29fa to
3bf3133
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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:
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/Avroclasses on the classpath.
In this initial version two table shapes are supported:
STATE_READER_MODE.KEYED): one row per key, with one column per registered keyed state (ValueStateas a scalar column,ListState/MapStateasARRAY/MAPcolumns).STATE_READER_MODE.KEYED_FLAT): a single namedListState/MapStateexposed 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 mappingTypeSerializerSnapshottrees to Flink SQLLogicalType(viaSerializerSnapshotToLogicalTypeConverter), 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 directStateTableUtils/CatalogTableconstruction andStateCatalog+ SQL, acrossprimitive, 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:
Documentation
Was generative AI tooling used to co-author this PR?