This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit f7e8448a17abd87639aa48e07ed64b847e351537 Author: Gyula Fora <[email protected]> AuthorDate: Sun Jul 19 22:35:01 2026 +0200 [FLINK-40179][state-processor-api] Window table mapping related utilities Adds the schema-extraction and catalog-table-building utilities needed to expose namespaced (e.g. window-scoped) keyed state: getWindowKeyedStateSchema, getWindowStateCatalogTable, and getFlattenedWindowStateCatalogTable in StateTableUtils, mirroring the plain keyed state utilities added previously. These are not yet wired into the catalog or table factory. --- .../apache/flink/state/api/StateTableUtils.java | 146 +++++++++++++++------ 1 file changed, 106 insertions(+), 40 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java index b6a4eacba30..731c60c0fac 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java @@ -152,6 +152,20 @@ public final class StateTableUtils { return buildKeyedStateSchemaInfo(schemas, classified.voidNamespaceStates, null); } + /** + * Like {@link #getKeyedStateSchema}, but for the namespaced (e.g. window-scoped) states exposed + * by the {@code _windowed}/{@code _windowed_flat} tables. Empty if the operator has no + * namespaced state; see {@link #classifyStates} for how multiple namespace types are handled. + */ + public static KeyedStateSchemaInfo getWindowKeyedStateSchema( + CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { + OperatorState opState = findOperatorState(metadata, operatorId); + List<StateSchemaInfo> schemas = StateSchemaExtractor.extractSchema(opState); + ClassifiedStates classified = classifyStates(operatorId.toString(), schemas); + return buildKeyedStateSchemaInfo( + schemas, classified.windowNamespaceStates, classified.windowLogicalType); + } + private static KeyedStateSchemaInfo buildKeyedStateSchemaInfo( List<StateSchemaInfo> allSchemas, List<StateSchemaInfo> statesToInclude, @@ -237,8 +251,67 @@ public final class StateTableUtils { } /** - * Builds a {@link CatalogTable} representing all keyed states of an operator, or, when {@code - * windowType} is non-null, all namespaced (e.g. window-scoped) states of an operator. + * Builds a {@link CatalogTable} exposing a single keyed LIST or MAP state flattened into one + * row per list element / map entry, rather than one row per key. + * + * <p>The resulting table has 3 columns, with a composite primary key on {@code state_key} and + * the sub-key column (the {@code state_key} value repeats across rows belonging to the same + * key, but the pair uniquely identifies a row). The third column has a fixed name — not the + * state's own name, to avoid collisions with other (reserved) column names: + * + * <ul> + * <li>LIST: {@code (state_key, list_index, list_value)}, primary key {@code (state_key, + * list_index)} + * <li>MAP: {@code (state_key, map_key, map_value)}, primary key {@code (state_key, map_key)} + * </ul> + * + * @param metadata the checkpoint metadata the operator belongs to + * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema} + * @param stateName the name of the LIST or MAP state to flatten + * @param statePath the path to the savepoint / checkpoint + * @param operatorIdentifier identifies the operator whose state to read + * @return a {@link CatalogTable} ready for registration + */ + public static CatalogTable getFlattenedStateCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + return buildFlattenedKeyedCatalogTable( + metadata, schemaInfo, stateName, statePath, operatorIdentifier, false); + } + + /** + * Builds a {@link CatalogTable} representing all namespaced (e.g. window-scoped) states of an + * operator, as returned by {@link #getWindowKeyedStateSchema}: one row per {@code (state_key, + * state_window)} pair, plus one column per namespaced VALUE-shaped state. + */ + public static CatalogTable getWindowStateCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String statePath, + OperatorIdentifier operatorIdentifier) { + + LogicalType windowType = + schemaInfo.stateSchemas.values().stream() + .map(entry -> entry.windowLogicalType) + .filter(java.util.Objects::nonNull) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "No namespaced state found for operator '" + + operatorIdentifier + + "'.")); + + return buildKeyedCatalogTable( + metadata, schemaInfo, statePath, operatorIdentifier, windowType); + } + + /** + * Builds a {@link CatalogTable} for either the plain keyed table ({@code windowType == null}) + * or the namespaced/window table (see {@link #getWindowStateCatalogTable}). */ private static CatalogTable buildKeyedCatalogTable( CheckpointMetadata metadata, @@ -261,57 +334,43 @@ public final class StateTableUtils { } if (windowType == null) { schemaBuilder.primaryKeyNamed("PK_state_key", "state_key"); + } else { + schemaBuilder.primaryKeyNamed("PK_state_key_state_window", "state_key", "state_window"); } Schema schema = schemaBuilder.build(); - Map<String, String> options = buildBaseConnectorOptions(statePath, operatorIdentifier); - options.put( - SavepointConnectorOptions.STATE_READER_MODE.key(), - (windowType == null + Map<String, String> options = + buildBaseConnectorOptions( + statePath, + operatorIdentifier, + windowType == null ? SavepointConnectorOptions.StateReaderMode.KEYED - : SavepointConnectorOptions.StateReaderMode.WINDOWED) - .toString()); + : SavepointConnectorOptions.StateReaderMode.WINDOWED); withStateBackendType(options, metadata, operatorIdentifier); return CatalogTable.newBuilder().schema(schema).options(options).build(); } /** - * Builds a {@link CatalogTable} exposing a single keyed LIST or MAP state flattened into one - * row per list element / map entry, rather than one row per key. - * - * <p>The resulting table has 3 columns, with a composite primary key on {@code state_key} and - * the sub-key column (the {@code state_key} value repeats across rows belonging to the same - * key, but the pair uniquely identifies a row). The third column has a fixed name — not the - * state's own name, to avoid collisions with other (reserved) column names: - * - * <ul> - * <li>LIST: {@code (state_key, list_index, list_value)}, primary key {@code (state_key, - * list_index)} - * <li>MAP: {@code (state_key, map_key, map_value)}, primary key {@code (state_key, map_key)} - * </ul> - * - * @param metadata the checkpoint metadata the operator belongs to - * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema} - * @param stateName the name of the LIST or MAP state to flatten - * @param statePath the path to the savepoint / checkpoint - * @param operatorIdentifier identifies the operator whose state to read - * @return a {@link CatalogTable} ready for registration + * Like {@link #getFlattenedStateCatalogTable}, but for a namespaced (window-scoped) LIST/MAP + * state: {@code (state_key, state_window, list_index, list_value)} for LIST, or {@code + * (state_key, state_window, map_key, map_value)} for MAP. */ - public static CatalogTable getFlattenedStateCatalogTable( + public static CatalogTable getFlattenedWindowStateCatalogTable( CheckpointMetadata metadata, KeyedStateSchemaInfo schemaInfo, String stateName, String statePath, OperatorIdentifier operatorIdentifier) { return buildFlattenedKeyedCatalogTable( - metadata, schemaInfo, stateName, statePath, operatorIdentifier, false); + metadata, schemaInfo, stateName, statePath, operatorIdentifier, true); } /** * Builds a {@link CatalogTable} exposing a single LIST or MAP state flattened into one row per * list element / map entry, either plain-keyed ({@code windowed == false}, see {@link - * #getFlattenedStateCatalogTable}) or namespaced ({@code windowed == true}). + * #getFlattenedStateCatalogTable}) or namespaced ({@code windowed == true}, see {@link + * #getFlattenedWindowStateCatalogTable}). */ private static CatalogTable buildFlattenedKeyedCatalogTable( CheckpointMetadata metadata, @@ -364,13 +423,13 @@ public final class StateTableUtils { } Schema schema = schemaBuilder.build(); - Map<String, String> options = buildBaseConnectorOptions(statePath, operatorIdentifier); - options.put( - SavepointConnectorOptions.STATE_READER_MODE.key(), - (windowed + Map<String, String> options = + buildBaseConnectorOptions( + statePath, + operatorIdentifier, + windowed ? SavepointConnectorOptions.StateReaderMode.WINDOWED_FLAT - : SavepointConnectorOptions.StateReaderMode.KEYED_FLAT) - .toString()); + : SavepointConnectorOptions.StateReaderMode.KEYED_FLAT); options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); withStateBackendType(options, metadata, operatorIdentifier); @@ -439,14 +498,21 @@ public final class StateTableUtils { /** * Returns the base connector options ({@link FactoryUtil#CONNECTOR}, {@link - * SavepointConnectorOptions#STATE_PATH}, and the operator identifier option) shared by every - * savepoint-backed {@link CatalogTable}. + * SavepointConnectorOptions#STATE_PATH}, {@link SavepointConnectorOptions#STATE_READER_MODE}, + * and the operator identifier option) shared by every savepoint-backed {@link CatalogTable}. + * + * <p>{@code readerMode} is always set explicitly rather than relying on {@link + * SavepointConnectorOptions#STATE_READER_MODE}'s default value, so that every catalog table's + * options unambiguously reflect its schema. */ private static Map<String, String> buildBaseConnectorOptions( - String statePath, OperatorIdentifier operatorIdentifier) { + String statePath, + OperatorIdentifier operatorIdentifier, + SavepointConnectorOptions.StateReaderMode readerMode) { Map<String, String> options = new HashMap<>(); options.put(FactoryUtil.CONNECTOR.key(), "savepoint"); options.put(SavepointConnectorOptions.STATE_PATH.key(), statePath); + options.put(SavepointConnectorOptions.STATE_READER_MODE.key(), readerMode.toString()); operatorIdentifier .getUid() .ifPresentOrElse(
