Conversation
…tilities Adds the schema-extraction and catalog-table-building utilities needed to expose an operator's non-keyed state (ListState/UnionState/ BroadcastState): NonKeyedStateSchemaInfo and OperatorStateSchemaInfo, plus hasNonInternalOperatorState, getNonKeyedStateSchema, getOperatorStateCatalogTable, and getBroadcastStateCatalogTable in StateTableUtils, mirroring the keyed/window state utilities added previously. Also extends getOperatorIdentifiers to include operators that register only non-keyed state. These are not yet wired into the catalog or table factory.
…g and factories Wires the non-keyed (ListState/UnionState/BroadcastState) utilities added previously into the catalog and table factory: - StateCatalog: adds the LIST/UNION/BROADCAST table suffixes and getTable() dispatch, and extends candidateTablesForOperator to enumerate one table per non-keyed state (an operator may register only non-keyed state, with no plain per-key "_keyed" table). - SavepointDynamicTableSourceFactory: adds the LIST/UNION/BROADCAST dispatch and createOperatorStateDynamicTableSource/ createBroadcastStateDynamicTableSource, mirroring the keyed/window table source creation. - Adds the DynamicTableSource/ScanProvider/TableMapping implementations for operator (list/union) and broadcast state: AbstractNonKeyedDynamicTableSource, AbstractNonKeyedDataStreamScanProvider, OperatorStateDynamicTableSource, OperatorStateDataStreamScanProvider, OperatorStateTableMapping, BroadcastStateDynamicTableSource, BroadcastStateDataStreamScanProvider, BroadcastStateTableMapping.
Adds StateCatalogNonKeyedITCase covering read access to operator ListState/UnionState/BroadcastState through the StateCatalog.
Collaborator
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
2 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.
What is the purpose of the change
The state catalog (introduced in FLINK-40176) lets you query savepoint/checkpoint state via SQL, but so far only covered keyed state. This PR extends it to non-keyed (operator) state —
ListState,UnionState, andBroadcastState— exposing each as its own table.Brief change log
CatalogTables forListState/UnionState(one row per element) andBroadcastState(one row per map entry, keyed onmap_key)StateCatalog(naming/listing) andSavepointDynamicTableSourceFactory(scan)Verifying this change
This change added tests and can be verified as follows:
StateCatalogNonKeyedITCase(run against both HashMap and RocksDB state backends), covering all three state shapes and the missing-POJO-class fallback pathDoes this pull request potentially affect one of the following parts:
@Public(Evolving): yes (StateCatalog, still@PublicEvolving)Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Sonnet 5)