From a5e541dd9ac49614d41dc2eb623d143368800bd8 Mon Sep 17 00:00:00 2001 From: Gyula Fora Date: Mon, 20 Jul 2026 08:48:03 +0200 Subject: [PATCH 1/3] [FLINK-40180][state-processor-api] Non-keyed/operator state related utilities 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. --- .../flink/state/api/StateTableUtils.java | 311 ++++++++++++------ .../api/schema/NonKeyedStateSchemaInfo.java | 68 ++++ .../api/schema/OperatorStateSchemaInfo.java | 66 ++++ .../api/schema/StateSchemaExtractor.java | 107 ++++-- 4 files changed, 421 insertions(+), 131 deletions(-) create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java 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 731c60c0fac046..9bf12c20fbfc3d 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 @@ -32,6 +32,8 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer; import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle; import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.NonKeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.OperatorStateSchemaInfo; import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter; import org.apache.flink.state.api.schema.StateSchemaExtractor; import org.apache.flink.state.api.schema.StateSchemaInfo; @@ -46,6 +48,7 @@ import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter; @@ -67,8 +70,8 @@ import java.util.stream.Collectors; /** - * High-level utility for inspecting and reading keyed state from a checkpoint / savepoint without - * requiring user POJO classes on the classpath. + * High-level utility for inspecting and reading keyed and non-keyed state from a checkpoint / + * savepoint without requiring user POJO classes on the classpath. */ @Internal public final class StateTableUtils { @@ -78,15 +81,11 @@ public final class StateTableUtils { private StateTableUtils() {} /** - * Returns the {@link OperatorIdentifier}s of all operators present in the given checkpoint - * metadata that have at least one non-internal keyed state. - * - * @param metadata the checkpoint metadata to inspect - * @return list of operator identifiers; never null, may be empty + * Returns identifiers of all operators with at least one non-internal keyed or non-keyed state. */ public static List getOperatorIdentifiers(CheckpointMetadata metadata) { return metadata.getOperatorStates().stream() - .filter(StateTableUtils::hasNonInternalKeyedState) + .filter(op -> hasNonInternalKeyedState(op) || hasNonInternalOperatorState(op)) .map( op -> op.getOperatorUid() @@ -113,15 +112,21 @@ private static boolean hasNonInternalKeyedState(OperatorState op) { } } - /** - * Returns the names of all keyed states registered by the given operator. - * - * @param metadata the checkpoint metadata to inspect - * @param operatorId identifies the operator - * @param classLoader the class loader used when reading serializer snapshots - * @return list of state names; never null, may be empty - * @throws IOException if the state header cannot be read - */ + private static boolean hasNonInternalOperatorState(OperatorState op) { + try { + List schemas = StateSchemaExtractor.extractOperatorSchema(op); + return schemas.stream().anyMatch(info -> !isInternalState(info.stateName)); + } catch (Exception e) { + LOG.error( + "Could not extract non-keyed state schema for operator '{}': {}. " + + "Excluding from catalog.", + op.getOperatorID(), + e.getMessage()); + return false; + } + } + + /** Returns the names of all keyed states registered by the given operator. */ public static List getKeyedStates( CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { OperatorState opState = findOperatorState(metadata, operatorId); @@ -133,16 +138,10 @@ public static List getKeyedStates( } /** - * Returns the {@link KeyedStateSchemaInfo} for the plain per-key (void-namespace) states of the - * given operator — the ones exposed by the {@code _keyed}/{@code _keyed_flat} tables. - * - *

Schema extraction is lenient: POJO field names and types are derived from the serializer - * snapshot and do not require the user POJO class to be on the classpath. - * - * @param metadata the checkpoint metadata to inspect - * @param operatorId identifies the operator - * @return schema information covering the key type and all registered state entries - * @throws IOException if the state header cannot be read + * Returns schema info for the plain per-key (void-namespace) states of the given operator — the + * ones exposed by the {@code _keyed}/{@code _keyed_flat} tables. Extraction is lenient: field + * names/types come from the serializer snapshot, so the user POJO class need not be on the + * classpath. */ public static KeyedStateSchemaInfo getKeyedStateSchema( CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { @@ -204,11 +203,7 @@ private static KeyedStateSchemaInfo buildKeyedStateSchemaInfo( } /** - * Logs that a single state's schema could not be extracted and will therefore be excluded from - * the table schema, shared by {@link #buildKeyedStateSchemaInfo}. - * - * @param label a prefix inserted before "state" in the log message (e.g. {@code "non-keyed "} - * or {@code ""}), distinguishing which caller excluded the state + * Logs that a state's schema extraction failed and it will be excluded from the table schema. */ private static void logSchemaExtractionFailure( String label, @@ -226,21 +221,10 @@ private static void logSchemaExtractionFailure( } /** - * Builds a {@link CatalogTable} representing all keyed states of an operator. - * - *

The resulting table has one column named {@code "state_key"} for the key and one column - * per keyed state. The connector options are pre-populated so the table can be registered - * directly in a {@link org.apache.flink.table.catalog.CatalogManager}. - * - *

When the state backend that produced the operator's keyed state can be unambiguously - * determined from the checkpoint metadata, {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} - * is pre-populated as well, so callers don't need to specify it themselves. - * - * @param metadata the checkpoint metadata the operator belongs to - * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema} - * @param statePath the path to the savepoint / checkpoint - * @param operatorIdentifier identifies the operator whose state to read - * @return a {@link CatalogTable} ready for registration + * Builds a {@link CatalogTable} for all keyed states of an operator: one {@code "state_key"} + * column plus one column per keyed state. Pre-populates {@link + * SavepointConnectorOptions#STATE_BACKEND_TYPE} when it can be unambiguously determined from + * the checkpoint metadata. */ public static CatalogTable getStateCatalogTable( CheckpointMetadata metadata, @@ -252,25 +236,13 @@ public static CatalogTable getStateCatalogTable( /** * 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. - * - *

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: + * row per list element / map entry, rather than one row per key. Value column names are fixed + * rather than the state's own name to avoid collisions with reserved columns: * *

    - *
  • LIST: {@code (state_key, list_index, list_value)}, primary key {@code (state_key, - * list_index)} - *
  • MAP: {@code (state_key, map_key, map_value)}, primary key {@code (state_key, map_key)} + *
  • LIST: {@code (state_key, list_index, list_value)}, PK {@code (state_key, list_index)} + *
  • MAP: {@code (state_key, map_key, map_value)}, PK {@code (state_key, map_key)} *
- * - * @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, @@ -466,16 +438,161 @@ private static String addFlattenedValueColumns( return subKeyColumnName; } + /** + * Returns schema info for the non-keyed (operator) states of the given operator — {@code + * ListState}, {@code UnionState}, {@code BroadcastState} — exposed by the {@code _list}/{@code + * _union}/{@code _broadcast} tables. Extraction is lenient (no user POJO class required on the + * classpath); states whose schema can't be determined are excluded with a logged error. + */ + public static NonKeyedStateSchemaInfo getNonKeyedStateSchema( + CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { + OperatorState opState = findOperatorState(metadata, operatorId); + List schemas = StateSchemaExtractor.extractOperatorSchema(opState); + + LinkedHashMap stateSchemas = + new LinkedHashMap<>(); + for (OperatorStateSchemaInfo info : schemas) { + if (isInternalState(info.stateName)) { + continue; + } + try { + LogicalType valueLogicalType = + SerializerSnapshotToLogicalTypeConverter.convert(info.valueSnapshot); + LogicalType mapKeyLogicalType = + info.keySnapshot == null + ? null + : SerializerSnapshotToLogicalTypeConverter.convert( + info.keySnapshot); + stateSchemas.put( + info.stateName, + new NonKeyedStateSchemaInfo.StateEntryInfo( + info.kind, valueLogicalType, mapKeyLogicalType)); + } catch (Exception e) { + logSchemaExtractionFailure("non-keyed ", info.stateName, info.valueSnapshot, e); + } + } + + return new NonKeyedStateSchemaInfo(stateSchemas); + } + + /** + * Builds a {@link CatalogTable} exposing a single {@code ListState} or {@code UnionState}, with + * one row per list element: a ROW-typed value contributes one column per field, a scalar value + * gets a single column named after the state. No ordering column and no primary key, since no + * column is guaranteed unique across rows. Unlike the keyed table builders, this does not set + * {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} since non-keyed state isn't part of a + * state backend. + */ + public static CatalogTable getOperatorStateCatalogTable( + NonKeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + + NonKeyedStateSchemaInfo.StateEntryInfo entryInfo = + findNonKeyedStateEntry(schemaInfo, stateName, operatorIdentifier); + if (entryInfo.kind != SavepointConnectorOptions.StateReaderMode.LIST + && entryInfo.kind != SavepointConnectorOptions.StateReaderMode.UNION) { + throw new IllegalArgumentException( + "Operator state tables are only supported for LIST and UNION states, but '" + + stateName + + "' is " + + entryInfo.kind + + "."); + } + + Schema.Builder schemaBuilder = Schema.newBuilder(); + if (entryInfo.valueLogicalType instanceof RowType) { + for (RowType.RowField field : ((RowType) entryInfo.valueLogicalType).getFields()) { + schemaBuilder.column( + field.getName(), LogicalTypeDataTypeConverter.toDataType(field.getType())); + } + } else { + schemaBuilder.column( + stateName, LogicalTypeDataTypeConverter.toDataType(entryInfo.valueLogicalType)); + } + + Map options = + buildBaseConnectorOptions(statePath, operatorIdentifier, entryInfo.kind); + options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); + + return CatalogTable.newBuilder().schema(schemaBuilder.build()).options(options).build(); + } + + /** + * Builds a {@link CatalogTable} exposing a single {@code BroadcastState}, one row per map + * entry: {@code (map_key NOT NULL, map_value)} with PK {@code map_key}. Column names are fixed + * rather than the state's own name to avoid collisions with reserved columns. Like {@link + * #getOperatorStateCatalogTable}, this does not set {@link + * SavepointConnectorOptions#STATE_BACKEND_TYPE}. + */ + public static CatalogTable getBroadcastStateCatalogTable( + NonKeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + + NonKeyedStateSchemaInfo.StateEntryInfo entryInfo = + findNonKeyedStateEntry(schemaInfo, stateName, operatorIdentifier); + if (entryInfo.kind != SavepointConnectorOptions.StateReaderMode.BROADCAST) { + throw new IllegalArgumentException( + "Broadcast state tables are only supported for BROADCAST states, but '" + + stateName + + "' is " + + entryInfo.kind + + "."); + } + + Schema schema = + Schema.newBuilder() + .column( + "map_key", + LogicalTypeDataTypeConverter.toDataType(entryInfo.mapKeyLogicalType) + .notNull()) + .column( + "map_value", + LogicalTypeDataTypeConverter.toDataType(entryInfo.valueLogicalType)) + .primaryKeyNamed("PK_map_key", "map_key") + .build(); + + Map options = + buildBaseConnectorOptions( + statePath, + operatorIdentifier, + SavepointConnectorOptions.StateReaderMode.BROADCAST); + options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); + + return CatalogTable.newBuilder().schema(schema).options(options).build(); + } + // ------------------------------------------------------------------------- // Private helpers // ------------------------------------------------------------------------- /** - * Resolves the SQL column {@link org.apache.flink.table.types.DataType} for a single state's - * value column, forcing it nullable for VALUE-shaped state: unlike LIST/MAP (which always have - * a value, possibly empty), a {@code ValueState}/{@code ReducingState}/{@code AggregatingState} - * can legitimately hold no value (e.g. never written, or cleared by a trigger such as {@code - * CountTrigger}), in which case a read returns {@code null}. + * Looks up a single non-keyed state entry, failing rather than returning {@code null} when the + * operator has no such state. + */ + private static NonKeyedStateSchemaInfo.StateEntryInfo findNonKeyedStateEntry( + NonKeyedStateSchemaInfo schemaInfo, + String stateName, + OperatorIdentifier operatorIdentifier) { + NonKeyedStateSchemaInfo.StateEntryInfo entryInfo = schemaInfo.stateSchemas.get(stateName); + if (entryInfo == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' not found for operator '" + + operatorIdentifier + + "'."); + } + return entryInfo; + } + + /** + * Resolves the value column's {@link org.apache.flink.table.types.DataType}, forcing it + * nullable for VALUE-shaped state: unlike LIST/MAP, a {@code ValueState} can legitimately hold + * no value (never written, or cleared by a trigger), so a read may return {@code null}. */ private static DataType stateValueColumnDataType( KeyedStateSchemaInfo.StateEntryInfo entryInfo) { @@ -497,13 +614,9 @@ private static OperatorState findOperatorState( } /** - * Returns the base connector options ({@link FactoryUtil#CONNECTOR}, {@link - * SavepointConnectorOptions#STATE_PATH}, {@link SavepointConnectorOptions#STATE_READER_MODE}, - * and the operator identifier option) shared by every savepoint-backed {@link CatalogTable}. - * - *

{@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. + * Returns the base connector options shared by every savepoint-backed {@link CatalogTable}. + * {@code readerMode} is always set explicitly rather than relying on the option's default, so + * every table's options unambiguously reflect its schema. */ private static Map buildBaseConnectorOptions( String statePath, @@ -525,10 +638,9 @@ private static Map buildBaseConnectorOptions( } /** - * Adds {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} to {@code options} when it can be - * unambiguously determined from the checkpoint metadata. Only meaningful for keyed state - * tables: non-keyed (list/union/broadcast) state isn't stored in a state backend, so callers - * for those table kinds must not call this. + * Adds {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} when it can be unambiguously + * determined. Only meaningful for keyed state tables — non-keyed state isn't stored in a state + * backend. */ private static void withStateBackendType( Map options, @@ -543,16 +655,12 @@ private static void withStateBackendType( } /** - * Attempts to determine the state backend (shortcut name, see {@link - * StateBackendLoader#HASHMAP_STATE_BACKEND_NAME} / {@link - * StateBackendLoader#ROCKSDB_STATE_BACKEND_NAME}) that produced the operator's keyed state, by - * inspecting the concrete {@link KeyedStateHandle} subtype found in the checkpoint metadata: - * heap/HashMap backends produce {@link KeyGroupsStateHandle}, RocksDB/ForSt backends produce - * {@link IncrementalKeyedStateHandle}. - * - *

Canonical-format savepoints rewrite keyed state into the backend-agnostic {@link - * KeyGroupsSavepointStateHandle}, in which case the originating backend can no longer be - * determined from the handle alone; an empty result is returned rather than guessing. + * Attempts to determine the state backend that produced the operator's keyed state, from the + * concrete {@link KeyedStateHandle} subtype: heap/HashMap backends produce {@link + * KeyGroupsStateHandle}, RocksDB/ForSt produce {@link IncrementalKeyedStateHandle}. + * Canonical-format savepoints rewrite state into the backend-agnostic {@link + * KeyGroupsSavepointStateHandle}, in which case the backend can't be told and an empty result + * is returned. */ static Optional detectStateBackendType(OperatorState opState) { Set detectedTypes = new HashSet<>(); @@ -584,8 +692,7 @@ private static void collectStateBackendTypes( } else if (handle instanceof IncrementalKeyedStateHandle) { detectedTypes.add(StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME); } else if (handle instanceof KeyGroupsSavepointStateHandle) { - // Canonical-format savepoints rewrite keyed state into a backend-agnostic - // format; the originating backend can no longer be told apart from the handle. + // backend-agnostic format; origin can't be told apart from the handle } else if (handle instanceof KeyGroupsStateHandle) { detectedTypes.add(StateBackendLoader.HASHMAP_STATE_BACKEND_NAME); } else { @@ -601,11 +708,9 @@ private static boolean isInternalState(String stateName) { } /** - * Returns {@code true} if a state is plain per-key state (registered with {@code - * VoidNamespace}) and {@code false} if it is scoped by some other namespace (e.g. a window). - * - *

A missing namespace snapshot (e.g. from an older savepoint format) is treated as void, - * matching pre-existing behavior. + * Returns {@code true} for plain per-key state (registered with {@code VoidNamespace}), {@code + * false} if scoped by another namespace (e.g. a window). A missing snapshot (older savepoint + * format) is treated as void. */ private static boolean isVoidNamespace(TypeSerializerSnapshot namespaceSnapshot) { return namespaceSnapshot == null @@ -633,16 +738,10 @@ private static final class ClassifiedStates { } /** - * Partitions the user-registered states of a single operator into plain per-key - * (void-namespace) states and namespaced states, resolving the namespaced states' shared {@link - * LogicalType} along the way. - * - *

An operator may register states under more than one distinct namespace type only - * via hand-rolled state access (no built-in windowing API does this); when that happens, the - * first namespace type whose schema can be determined is kept, and every other group is - * excluded with a logged warning. - * - * @param operatorLabel a human-readable operator identifier, used only for log messages + * Partitions user-registered states into plain per-key (void-namespace) and namespaced states, + * resolving the namespaced states' shared {@link LogicalType}. An operator can register more + * than one namespace type only via hand-rolled state access; when that happens, the first + * resolvable namespace type is kept and the rest excluded with a warning. */ private static ClassifiedStates classifyStates( String operatorLabel, List schemas) { diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java new file mode 100644 index 00000000000000..8f33bb4ecb0ae1 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; +import org.apache.flink.table.types.logical.LogicalType; + +import javax.annotation.Nullable; + +import java.util.LinkedHashMap; + +/** + * Schema information for all non-keyed (operator) states of a single operator — {@code ListState}, + * {@code UnionState}, and {@code BroadcastState} — extracted from a savepoint without requiring the + * user POJO classes on the classpath. + * + *

Unlike {@link KeyedStateSchemaInfo}, there is no shared key type across entries: operator + * state has no per-key partitioning, and only {@link StateReaderMode#BROADCAST} entries carry a + * (per-entry) map key type. + */ +@Internal +public final class NonKeyedStateSchemaInfo { + + /** State names to entry info, in savepoint registration order. */ + public final LinkedHashMap stateSchemas; + + public NonKeyedStateSchemaInfo(LinkedHashMap stateSchemas) { + this.stateSchemas = stateSchemas; + } + + /** Schema information for one non-keyed state entry. */ + public static final class StateEntryInfo { + + public final StateReaderMode kind; + + /** Logical type of the state's value (list element type, or broadcast map's value type). */ + public final LogicalType valueLogicalType; + + /** Broadcast map's key type; non-null only for {@link StateReaderMode#BROADCAST}. */ + @Nullable public final LogicalType mapKeyLogicalType; + + public StateEntryInfo( + StateReaderMode kind, + LogicalType valueLogicalType, + @Nullable LogicalType mapKeyLogicalType) { + this.kind = kind; + this.valueLogicalType = valueLogicalType; + this.mapKeyLogicalType = mapKeyLogicalType; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java new file mode 100644 index 00000000000000..3100415186304f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; + +import javax.annotation.Nullable; + +/** + * Carries the schema information extracted from a single non-keyed (operator) state entry — {@code + * ListState}, {@code UnionState}, or {@code BroadcastState} — in a savepoint, without requiring the + * user POJO classes on the classpath. + * + *

Unlike {@link StateSchemaInfo}, there is no per-key serializer: non-keyed state has no key + * concept. {@link #keySnapshot} is only present for {@link StateReaderMode#BROADCAST}, where it + * represents the broadcast map's key. + */ +@Internal +public final class OperatorStateSchemaInfo { + + /** Name of the state as registered by the operator. */ + public final String stateName; + + public final StateReaderMode kind; + + /** + * Serializer snapshot for the state's value type: the map's value type for {@link + * StateReaderMode#BROADCAST}, the list element type for {@link StateReaderMode#LIST}/{@link + * StateReaderMode#UNION}. + */ + public final TypeSerializerSnapshot valueSnapshot; + + /** + * Broadcast map's key serializer snapshot; non-null only for {@link StateReaderMode#BROADCAST}. + */ + @Nullable public final TypeSerializerSnapshot keySnapshot; + + public OperatorStateSchemaInfo( + String stateName, + StateReaderMode kind, + TypeSerializerSnapshot valueSnapshot, + @Nullable TypeSerializerSnapshot keySnapshot) { + this.stateName = stateName; + this.kind = kind; + this.valueSnapshot = valueSnapshot; + this.keySnapshot = keySnapshot; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java index c58c7bf77370b6..6774c0e5d845f4 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java @@ -29,23 +29,26 @@ import org.apache.flink.runtime.state.IncrementalKeyedStateHandle; import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.OperatorBackendSerializationProxy; +import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StreamStateHandle; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.CommonOptionsKeys; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.CommonSerializerKeys; import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory; +import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; import java.io.IOException; +import java.io.InputStream; import java.util.ArrayList; import java.util.Collections; import java.util.List; /** - * Utility for extracting {@link StateSchemaInfo} from a savepoint without instantiating the full - * state backend or requiring user POJO classes on the classpath. - * - *

It reads the {@link KeyedBackendSerializationProxy} header that every heap/RocksDB keyed state - * file starts with. + * Utility for extracting {@link StateSchemaInfo} / {@link OperatorStateSchemaInfo} from a savepoint + * without instantiating the full state backend or requiring user POJO classes on the classpath, by + * reading the {@link KeyedBackendSerializationProxy} / {@link OperatorBackendSerializationProxy} + * header that every keyed / non-keyed state file starts with. */ @Internal public final class StateSchemaExtractor { @@ -53,23 +56,14 @@ public final class StateSchemaExtractor { private StateSchemaExtractor() {} /** - * Reads state schema information from the first available keyed state handle in the given - * operator state. - * - *

Returns an empty list rather than throwing when no keyed state handle is found: an - * operator may register only non-keyed (list/union/broadcast) state, in which case it has no - * keyed state to describe. + * Reads state schema info from the first available keyed state handle. Returns an empty list + * (rather than throwing) when the operator registers only non-keyed state. * - *

The metadata header lives in different places depending on the state backend: heap ({@code - * HashMapStateBackend}) savepoints hand back a {@code KeyGroupsStateHandle}, which is itself a - * {@link StreamStateHandle} starting with the header; RocksDB (incremental or full native) - * snapshots hand back an {@link IncrementalKeyedStateHandle}, whose own data stream starts with - * the SST payload instead, so the header must be read from {@link + *

The metadata header's location depends on the backend: heap ({@code HashMapStateBackend}) + * hands back a {@code KeyGroupsStateHandle} that is itself a {@link StreamStateHandle} starting + * with the header, while RocksDB hands back an {@link IncrementalKeyedStateHandle} whose stream + * starts with the SST payload instead, so the header must come from {@link * IncrementalKeyedStateHandle#getMetaDataStateHandle()}. - * - * @param operatorState the operator state from a loaded savepoint / checkpoint metadata - * @return list of schema info, one entry per registered state; never null, may be empty - * @throws IOException if the state header cannot be read */ public static List extractSchema(OperatorState operatorState) throws IOException { @@ -84,7 +78,7 @@ public static List extractSchema(OperatorState operatorState) metadataHandle = (StreamStateHandle) handle; } if (metadataHandle != null) { - try (java.io.InputStream stream = metadataHandle.openInputStream()) { + try (InputStream stream = metadataHandle.openInputStream()) { return extractSchema(new DataInputViewStreamWrapper(stream)); } } @@ -93,10 +87,7 @@ public static List extractSchema(OperatorState operatorState) return Collections.emptyList(); } - /** - * Package-private overload that accepts a {@link DataInputView} directly. Allows unit tests to - * inject pre-built byte arrays without a real filesystem. - */ + /** Package-private overload for tests to inject pre-built bytes without a real filesystem. */ static List extractSchema(DataInputView in) throws IOException { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); KeyedBackendSerializationProxy proxy = new KeyedBackendSerializationProxy<>(classLoader); @@ -138,4 +129,70 @@ static List extractSchema(DataInputView in) throws IOException return result; } + + /** + * Reads non-keyed (operator) state schema info — {@code ListState}, {@code UnionState}, {@code + * BroadcastState} — from the first available operator state handle. Returns an empty list + * (rather than throwing) when none is found. Unlike keyed state, an {@link OperatorStateHandle} + * is itself a {@link StreamStateHandle} starting with the metadata header, for every backend. + */ + public static List extractOperatorSchema(OperatorState operatorState) + throws IOException { + + for (OperatorSubtaskState subtask : operatorState.getSubtaskStates().values()) { + for (OperatorStateHandle handle : subtask.getManagedOperatorState()) { + try (InputStream stream = handle.openInputStream()) { + return extractOperatorSchema(new DataInputViewStreamWrapper(stream)); + } + } + } + return Collections.emptyList(); + } + + /** Package-private overload for tests to inject pre-built bytes without a real filesystem. */ + static List extractOperatorSchema(DataInputView in) + throws IOException { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + OperatorBackendSerializationProxy proxy = + new OperatorBackendSerializationProxy(classLoader); + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + proxy.read(in); + + List result = new ArrayList<>(); + + for (StateMetaInfoSnapshot meta : proxy.getOperatorStateMetaInfoSnapshots()) { + TypeSerializerSnapshot valueSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER); + if (valueSnapshot == null) { + continue; + } + + // Only union-distributed list state is redistributed as a whole on rescale; every + // other distribution mode behaves like a plain (split) ListState here. + String distributionMode = + meta.getOption(CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE); + StateReaderMode kind = + OperatorStateHandle.Mode.UNION.name().equals(distributionMode) + ? StateReaderMode.UNION + : StateReaderMode.LIST; + + result.add(new OperatorStateSchemaInfo(meta.getName(), kind, valueSnapshot, null)); + } + + for (StateMetaInfoSnapshot meta : proxy.getBroadcastStateMetaInfoSnapshots()) { + TypeSerializerSnapshot valueSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER); + TypeSerializerSnapshot keySnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.KEY_SERIALIZER); + if (valueSnapshot == null || keySnapshot == null) { + continue; + } + + result.add( + new OperatorStateSchemaInfo( + meta.getName(), StateReaderMode.BROADCAST, valueSnapshot, keySnapshot)); + } + + return result; + } } From 52685f0e566015ce6b7f30bc824a7b718785d755 Mon Sep 17 00:00:00 2001 From: Gyula Fora Date: Mon, 20 Jul 2026 09:02:31 +0200 Subject: [PATCH 2/3] [FLINK-40180][state-processor-api] Non-keyed/operator state in catalog 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. --- .../api/input/OperatorStateInputFormat.java | 7 + .../state/api/runtime/SavepointLoader.java | 109 +++++++++--- .../flink/state/catalog/StateCatalog.java | 64 +++++-- ...bstractNonKeyedDataStreamScanProvider.java | 106 +++++++++++ .../BroadcastStateDataStreamScanProvider.java | 109 ++++++++++++ .../table/BroadcastStateTableMapping.java | 166 ++++++++++++++++++ .../table/NonKeyedDynamicTableSource.java | 108 ++++++++++++ .../OperatorStateDataStreamScanProvider.java | 123 +++++++++++++ .../table/OperatorStateTableMapping.java | 145 +++++++++++++++ .../SavepointDynamicTableSourceFactory.java | 111 +++++++++++- .../state/table/TableMappingSupport.java | 15 ++ 11 files changed, 1023 insertions(+), 40 deletions(-) create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java index 65745a1b16ab55..c768f233fcf787 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java @@ -24,6 +24,7 @@ import org.apache.flink.api.common.io.DefaultInputSplitAssigner; import org.apache.flink.api.common.io.RichInputFormat; import org.apache.flink.api.common.io.statistics.BaseStatistics; +import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.core.io.InputSplitAssigner; @@ -36,6 +37,7 @@ import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory; import org.apache.flink.state.api.input.splits.OperatorStateInputSplit; import org.apache.flink.streaming.api.operators.StreamOperatorStateContext; import org.apache.flink.util.CollectionUtil; @@ -176,6 +178,11 @@ public void open(OperatorStateInputSplit split) throws IOException { ExecutionConfig executionConfig = KeyedStateInputFormat.deserialize( serializedExecutionConfig, runtimeContext.getUserCodeClassLoader()); + + // Falls back to RowData/GenericRecord for POJO/Avro state whose class is missing from the + // classpath. Must be set before the backend restores, which eagerly restores serializers. + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + final StreamOperatorStateContext context = new StreamOperatorContextBuilder( runtimeContext, diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java index b3ec1a0ce3cc19..4328388dda8734 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java @@ -31,6 +31,8 @@ import org.apache.flink.runtime.state.KeyGroupsStateHandle; import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.OperatorBackendSerializationProxy; +import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StreamStateHandle; import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; @@ -41,6 +43,7 @@ import java.io.DataInputStream; import java.io.IOException; +import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.stream.Collectors; @@ -73,6 +76,27 @@ public static final class OperatorStateMetadata { } } + /** + * Non-keyed (operator) state metadata loaded in a single I/O pass. List/union and broadcast + * states are kept in separate maps since a list/union state may share its name with a + * differently-typed broadcast state. + */ + public static final class NonKeyedOperatorStateMetadata { + + /** Per-state serializer snapshots for {@code ListState}/{@code UnionState}, by name. */ + public final Map operatorStateSnapshots; + + /** Per-state serializer snapshots for {@code BroadcastState}, by name. */ + public final Map broadcastStateSnapshots; + + NonKeyedOperatorStateMetadata( + Map operatorStateSnapshots, + Map broadcastStateSnapshots) { + this.operatorStateSnapshots = operatorStateSnapshots; + this.broadcastStateSnapshots = broadcastStateSnapshots; + } + } + /** * Takes the given string (representing a pointer to a checkpoint) and resolves it to a file * status for the checkpoint's metadata file. @@ -122,40 +146,65 @@ public static Map loadOperatorStateMetadata( public static OperatorStateMetadata loadOperatorMetadata( String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { - CheckpointMetadata checkpointMetadata = loadSavepointMetadata(savepointPath); + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); - OperatorState operatorState = - checkpointMetadata.getOperatorStates().stream() - .filter( - state -> - operatorIdentifier - .getOperatorId() - .equals(state.getOperatorID())) + KeyedStateHandle keyedStateHandle = + operatorState.getStates().stream() + .flatMap(s -> s.getManagedKeyedState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "Operator " - + operatorIdentifier - + " not found in savepoint")); + "No keyed state found for operator " + + operatorIdentifier)); - KeyedStateHandle keyedStateHandle = + KeyedBackendSerializationProxy proxy = readSerializationProxy(keyedStateHandle); + return new OperatorStateMetadata( + byStateName(proxy.getStateMetaInfoSnapshots()), proxy.getKeySerializerSnapshot()); + } + + /** + * Loads the per-state serializer snapshots of an operator's non-keyed (operator) state — {@code + * ListState}/{@code UnionState}/{@code BroadcastState} — in a single I/O operation. + */ + public static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); + + OperatorStateHandle operatorStateHandle = operatorState.getStates().stream() - .flatMap(s -> s.getManagedKeyedState().stream()) + .flatMap(s -> s.getManagedOperatorState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "No keyed state found for operator " + "No operator state found for operator " + operatorIdentifier)); - KeyedBackendSerializationProxy proxy = readSerializationProxy(keyedStateHandle); - Map stateSnapshots = - proxy.getStateMetaInfoSnapshots().stream() - .collect( - Collectors.toMap( - StateMetaInfoSnapshot::getName, Function.identity())); - return new OperatorStateMetadata(stateSnapshots, proxy.getKeySerializerSnapshot()); + OperatorBackendSerializationProxy proxy = readSerializationProxy(operatorStateHandle); + return new NonKeyedOperatorStateMetadata( + byStateName(proxy.getOperatorStateMetaInfoSnapshots()), + byStateName(proxy.getBroadcastStateMetaInfoSnapshots())); + } + + private static OperatorState findOperatorState( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + return loadSavepointMetadata(savepointPath).getOperatorStates().stream() + .filter(state -> operatorIdentifier.getOperatorId().equals(state.getOperatorID())) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Operator " + + operatorIdentifier + + " not found in savepoint")); + } + + private static Map byStateName( + List snapshots) { + return snapshots.stream() + .collect(Collectors.toMap(StateMetaInfoSnapshot::getName, Function.identity())); } private static KeyedBackendSerializationProxy readSerializationProxy( @@ -187,4 +236,22 @@ private static KeyedBackendSerializationProxy readSerializationProxy( return proxy; } } + + private static OperatorBackendSerializationProxy readSerializationProxy( + OperatorStateHandle stateHandle) throws IOException { + + // Unlike keyed state, an OperatorStateHandle is itself a StreamStateHandle whose stream + // starts with the metadata header, for every state backend. + try (FSDataInputStream inputStream = stateHandle.openInputStream()) { + DataInputViewStreamWrapper inputView = new DataInputViewStreamWrapper(inputStream); + + OperatorBackendSerializationProxy proxy = + new OperatorBackendSerializationProxy( + Thread.currentThread().getContextClassLoader()); + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + proxy.read(inputView); + + return proxy; + } + } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java index 06fab3674b18e2..290c1cce35ba21 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java @@ -24,6 +24,7 @@ import org.apache.flink.state.api.StateTableUtils; import org.apache.flink.state.api.runtime.SavepointLoader; import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.NonKeyedStateSchemaInfo; import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; import org.apache.flink.state.table.SavepointConnectorOptions.StateType; import org.apache.flink.table.api.DataTypes; @@ -76,8 +77,8 @@ *

    *
  • Catalog: the name given at registration time (e.g. {@code "state"}) *
  • Database: one entry per discovered snapshot (e.g. {@code "app1_savepoint-acce1cedsad"}) - *
  • Table: a single view named {@code "metadata"} per database, backed by the {@code - * savepoint_metadata} function from {@code StateModule} + *
  • Table: one per state exposed by the snapshot, plus a view named {@code "metadata"} per + * database, backed by the {@code savepoint_metadata} function from {@code StateModule} *
* *

Database names preserve hyphens from the original directory names. Backtick quoting is @@ -113,6 +114,9 @@ public class StateCatalog extends AbstractCatalog { public static final String FLAT_STATE_TABLE_SUFFIX = "_keyed_flat"; public static final String WINDOW_TABLE_SUFFIX = "_windowed"; public static final String FLAT_WINDOW_TABLE_SUFFIX = "_windowed_flat"; + public static final String LIST_TABLE_SUFFIX = "_list"; + public static final String UNION_TABLE_SUFFIX = "_union"; + public static final String BROADCAST_TABLE_SUFFIX = "_broadcast"; private static final CatalogDatabase EMPTY_DATABASE = new CatalogDatabaseImpl(Collections.emptyMap(), ""); @@ -327,6 +331,29 @@ public CatalogBaseTable getTable(ObjectPath tablePath) snapshotPath.get(), resolved.operatorIdentifier); } + case LIST: + case UNION: + { + NonKeyedStateSchemaInfo schemaInfo = + StateTableUtils.getNonKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getOperatorStateCatalogTable( + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } + case BROADCAST: + { + NonKeyedStateSchemaInfo schemaInfo = + StateTableUtils.getNonKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getBroadcastStateCatalogTable( + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } default: throw new IllegalStateException("Unhandled table kind " + resolved.kind); } @@ -620,10 +647,19 @@ private static CatalogView buildMetadataView(String snapshotPath) { // Operator table helpers // ------------------------------------------------------------------------- + private static final Map TABLE_SUFFIXES = + Map.of( + StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, + StateReaderMode.KEYED_FLAT, FLAT_STATE_TABLE_SUFFIX, + StateReaderMode.WINDOWED, WINDOW_TABLE_SUFFIX, + StateReaderMode.WINDOWED_FLAT, FLAT_WINDOW_TABLE_SUFFIX, + StateReaderMode.LIST, LIST_TABLE_SUFFIX, + StateReaderMode.UNION, UNION_TABLE_SUFFIX, + StateReaderMode.BROADCAST, BROADCAST_TABLE_SUFFIX); + /** * Table name for a {@code kind} of operator state, optionally scoped to one flattened/non-keyed - * state (see {@link #OPERATOR_TABLE_SUFFIX}/{@link #FLAT_STATE_TABLE_SUFFIX}/{@link - * #WINDOW_TABLE_SUFFIX}/{@link #FLAT_WINDOW_TABLE_SUFFIX}). + * state (see {@link #TABLE_SUFFIXES}). * *

{@code stateName} must be {@code null} for {@link StateReaderMode#KEYED}/{@link * StateReaderMode#WINDOWED} (the general keyed/namespaced table, one per operator) and non-null @@ -631,13 +667,6 @@ private static CatalogView buildMetadataView(String snapshotPath) { * — the state name alone disambiguates the table since keyed/non-keyed state names are unique * within an operator). */ - private static final Map TABLE_SUFFIXES = - Map.of( - StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, - StateReaderMode.KEYED_FLAT, FLAT_STATE_TABLE_SUFFIX, - StateReaderMode.WINDOWED, WINDOW_TABLE_SUFFIX, - StateReaderMode.WINDOWED_FLAT, FLAT_WINDOW_TABLE_SUFFIX); - static String tableName( OperatorIdentifier opId, StateReaderMode kind, @Nullable String stateName) { String base = @@ -702,8 +731,8 @@ private static Optional resolveTable( /** * Enumerates every table that {@code opId} contributes: the general keyed/window table (if any - * plain per-key/namespaced state is registered), plus one flattened table per LIST/MAP keyed or - * window state. + * plain per-key/namespaced state is registered), one flattened table per LIST/MAP keyed or + * window state, and one table per non-keyed (list/union/broadcast) state. * *

Shared by {@link #listTables} (which collects names for every candidate) and {@link * #resolveTable} (which matches candidates against a target name), so that adding a new state @@ -740,6 +769,15 @@ private static List candidateTablesForOperator( } } + // Non-keyed states each get their own table; the entry's kind already carries the table + // shape (LIST/UNION/BROADCAST), so there is no general per-operator table here. + NonKeyedStateSchemaInfo nonKeyedSchemaInfo = + StateTableUtils.getNonKeyedStateSchema(metadata, opId); + for (Map.Entry entry : + nonKeyedSchemaInfo.stateSchemas.entrySet()) { + candidates.add(new ResolvedTable(opId, entry.getValue().kind, entry.getKey())); + } + return candidates; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java new file mode 100644 index 00000000000000..69d29ba4818a80 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.connector.ProviderContext; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.utils.TypeConversions; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Shared scan-time logic for {@link OperatorStateDataStreamScanProvider} and {@link + * BroadcastStateDataStreamScanProvider}: opens the {@link SavepointReader} against the configured + * state backend and resolves the lazy mapping. Subclasses supply the actual {@link SavepointReader} + * call and its row-mapping logic via {@link #readState}. + * + *

Unlike keyed state (see {@link AbstractSavepointDataStreamScanProvider}), operator {@code + * ListState}/{@code UnionState}/{@code BroadcastState} have no key column. + */ +@Internal +abstract class AbstractNonKeyedDataStreamScanProvider implements DataStreamScanProvider { + + @Nullable protected final String stateBackendType; + protected final String statePath; + protected final OperatorIdentifier operatorIdentifier; + private final Supplier mappingSupplier; + protected final RowType rowType; + + protected AbstractNonKeyedDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier mappingSupplier, + final RowType rowType) { + this.stateBackendType = stateBackendType; + this.statePath = statePath; + this.operatorIdentifier = operatorIdentifier; + this.mappingSupplier = mappingSupplier; + this.rowType = rowType; + } + + @Override + public boolean isBounded() { + return true; + } + + @Override + public DataStream produceDataStream( + ProviderContext providerContext, StreamExecutionEnvironment execEnv) { + try { + SavepointReader savepointReader = + AbstractSavepointDataStreamScanProvider.createSavepointReader( + stateBackendType, statePath, execEnv, getClass().getClassLoader()); + + // Resolve the lazy mapping at scan time (class loading deferred from planning). + return readState(savepointReader, mappingSupplier.get()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** + * Drives the actual {@link SavepointReader} call for this scan and maps the result into {@link + * #rowType}-shaped rows. + */ + protected abstract DataStream readState(SavepointReader savepointReader, M mapping) + throws Exception; + + /** + * The {@link TypeInformation} the {@link SavepointReader} needs for a raw state value: its SQL + * logical type paired with the serializer the state was written with. + */ + static TypeInformation externalTypeInfo( + LogicalType logicalType, TypeSerializer serializer) { + return ExternalTypeInfo.of(TypeConversions.fromLogicalToDataType(logicalType), serializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java new file mode 100644 index 00000000000000..7163ce44e7f7a4 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for an operator {@code BroadcastState} table, exposing one + * row per map entry as {@code (map_key, value)}. + */ +@Internal +public class BroadcastStateDataStreamScanProvider + extends AbstractNonKeyedDataStreamScanProvider { + + public BroadcastStateDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier mappingSupplier, + final RowType rowType) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream readState( + SavepointReader savepointReader, BroadcastStateTableMapping mapping) throws Exception { + LogicalType mapKeyLogicalType = + rowType.getFields().get(BroadcastStateTableMapping.MAP_KEY_COLUMN_INDEX).getType(); + LogicalType valueLogicalType = + rowType.getFields().get(BroadcastStateTableMapping.VALUE_COLUMN_INDEX).getType(); + + TypeSerializer mapKeySerializer = + (TypeSerializer) mapping.getMapKeyTypeSerializer(); + TypeSerializer valueSerializer = + (TypeSerializer) mapping.getValueTypeSerializer(); + + DataStream> raw = + savepointReader.readBroadcastState( + operatorIdentifier, + mapping.getStateName(), + externalTypeInfo(mapKeyLogicalType, mapKeySerializer), + externalTypeInfo(valueLogicalType, valueSerializer), + mapKeySerializer, + valueSerializer); + + return raw.map(new BroadcastStateRowMapper(mapKeyLogicalType, valueLogicalType)) + .returns(InternalTypeInfo.of(rowType)); + } + + /** Converts a raw {@code (map_key, value)} tuple into its {@link RowData} representation. */ + private static class BroadcastStateRowMapper + implements MapFunction, RowData> { + + private final LogicalType mapKeyLogicalType; + private final LogicalType valueLogicalType; + private final StateValueConverter converter = new StateValueConverter(); + + private BroadcastStateRowMapper( + LogicalType mapKeyLogicalType, LogicalType valueLogicalType) { + this.mapKeyLogicalType = mapKeyLogicalType; + this.valueLogicalType = valueLogicalType; + } + + @Override + public RowData map(Tuple2 entry) { + GenericRowData row = new GenericRowData(RowKind.INSERT, 2); + row.setField( + BroadcastStateTableMapping.MAP_KEY_COLUMN_INDEX, + converter.getValue(mapKeyLogicalType, entry.f0)); + row.setField( + BroadcastStateTableMapping.VALUE_COLUMN_INDEX, + converter.getValue(valueLogicalType, entry.f1)); + return row; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java new file mode 100644 index 00000000000000..d1ee179b4a6bac --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import java.io.Serializable; +import java.util.List; + +/** + * Maps the fixed 2-column schema of a {@code BroadcastState} table: {@code (map_key NOT + * NULL, map_value)}, with a primary key on {@code map_key}. + * + *

The second column has a fixed name ({@code map_value}) rather than being named after the state + * itself, to avoid collisions with other (reserved) column names; the true state name is instead + * resolved from {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}. + * + *

Both the map key and value serializers are stored flat/unwrapped in the savepoint metadata + * (under {@code KEY_SERIALIZER}/{@code VALUE_SERIALIZER}) — unlike a keyed {@code MapState}, there + * is no wrapping {@code MapSerializer} to unwrap. + */ +@Internal +public class BroadcastStateTableMapping implements Serializable { + + private static final long serialVersionUID = 1L; + + public static final int MAP_KEY_COLUMN_INDEX = 0; + public static final int VALUE_COLUMN_INDEX = 1; + + private final String stateName; + private final TypeSerializer mapKeyTypeSerializer; + private final TypeSerializer valueTypeSerializer; + + public BroadcastStateTableMapping( + String stateName, + TypeSerializer mapKeyTypeSerializer, + TypeSerializer valueTypeSerializer) { + this.stateName = stateName; + this.mapKeyTypeSerializer = mapKeyTypeSerializer; + this.valueTypeSerializer = valueTypeSerializer; + } + + public String getStateName() { + return stateName; + } + + public TypeSerializer getMapKeyTypeSerializer() { + return mapKeyTypeSerializer; + } + + public TypeSerializer getValueTypeSerializer() { + return valueTypeSerializer; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema matches the fixed 2-column {@code (map_key, map_value)} + * layout with a primary key on {@code map_key}. This is a purely structural check; it performs + * no I/O or class loading. + */ + public static void validateSchema(ResolvedCatalogTable catalogTable) { + ResolvedSchema schema = catalogTable.getResolvedSchema(); + List columns = schema.getColumns(); + if (columns.size() != 2) { + throw new ValidationException( + "BROADCAST state tables must have exactly 2 columns " + + "(map_key, map_value), but found " + + columns.size() + + "."); + } + String mapKeyColumnName = columns.get(MAP_KEY_COLUMN_INDEX).getName(); + if (!"map_key".equals(mapKeyColumnName)) { + throw new ValidationException( + "BROADCAST state tables must name their first column 'map_key', " + + "but found '" + + mapKeyColumnName + + "'."); + } + + String valueColumnName = columns.get(VALUE_COLUMN_INDEX).getName(); + if (!"map_value".equals(valueColumnName)) { + throw new ValidationException( + "BROADCAST state tables must name their second column 'map_value', " + + "but found '" + + valueColumnName + + "'."); + } + + List primaryKeyColumns = + schema.getPrimaryKey().map(UniqueConstraint::getColumns).orElse(List.of()); + if (!primaryKeyColumns.equals(List.of(mapKeyColumnName))) { + throw new ValidationException( + "BROADCAST state tables must declare a primary key on '" + + mapKeyColumnName + + "', but found: " + + (primaryKeyColumns.isEmpty() ? "none" : primaryKeyColumns) + + "."); + } + } + + /** + * Builds a complete {@link BroadcastStateTableMapping}, loading non-keyed operator state + * metadata from the savepoint and resolving the key and value serializers from it. + * + *

Assumes {@link #validateSchema} has already been called. This performs I/O; callers should + * invoke it lazily, deferred to scan time, to keep planning free of savepoint access. + */ + public static BroadcastStateTableMapping from( + ResolvedCatalogTable catalogTable, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig) { + + NonKeyedOperatorStateMetadata operatorMetadata = + TableMappingSupport.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + + SavepointTypeInfoResolver typeResolver = + new SavepointTypeInfoResolver( + operatorMetadata.broadcastStateSnapshots, serializerConfig, null); + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + RowType rowType = (RowType) physicalDataType.getLogicalType(); + LogicalType valueLogicalType = rowType.getFields().get(VALUE_COLUMN_INDEX).getType(); + // Synthetic RowField: name == actual state name (so metadata lookup succeeds), since the + // value column itself is named after a fixed literal (map_value), not the state. + RowType.RowField valueRowField = new RowType.RowField(stateName, valueLogicalType); + + TypeSerializer mapKeyTypeSerializer = typeResolver.resolveKeySerializer(valueRowField); + TypeSerializer valueTypeSerializer = + typeResolver.resolveFlatValueSerializer(valueRowField); + + return new BroadcastStateTableMapping(stateName, mapKeyTypeSerializer, valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java new file mode 100644 index 00000000000000..149deabb0e0444 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Dynamic table source for a single non-keyed operator state: {@code ListState}/{@code UnionState} + * (value flattened into the table's columns) or {@code BroadcastState} (fixed {@code (map_key, + * map_value)} schema). + * + *

Unlike keyed state (see {@link AbstractSavepointDynamicTableSource}), non-keyed state has no + * key column, so neither filter nor projection push-down is supported. The scan is delegated to the + * {@link DataStreamScanProvider} supplied by {@link SavepointDynamicTableSourceFactory}, so a + * single table-source class serves every non-keyed state kind. + */ +@Internal +public class NonKeyedDynamicTableSource implements ScanTableSource { + + /** Builds the {@link DataStreamScanProvider} for a given set of scan-time arguments. */ + interface ScanProviderFactory { + DataStreamScanProvider create( + @Nullable String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier, + Supplier mappingSupplier, + RowType rowType); + } + + @Nullable private final String stateBackendType; + private final String statePath; + private final OperatorIdentifier operatorIdentifier; + private final Supplier mappingSupplier; + private final RowType rowType; + private final String summaryString; + private final ScanProviderFactory scanProviderFactory; + + public NonKeyedDynamicTableSource( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier mappingSupplier, + final RowType rowType, + final String summaryString, + final ScanProviderFactory scanProviderFactory) { + this.stateBackendType = stateBackendType; + this.statePath = statePath; + this.operatorIdentifier = operatorIdentifier; + this.mappingSupplier = mappingSupplier; + this.rowType = rowType; + this.summaryString = summaryString; + this.scanProviderFactory = scanProviderFactory; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { + return scanProviderFactory.create( + stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + public DynamicTableSource copy() { + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + summaryString, + scanProviderFactory); + } + + @Override + public String asSummaryString() { + return summaryString; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java new file mode 100644 index 00000000000000..3c90b35be1f1b1 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for an operator {@code ListState}/{@code UnionState} table. + * + *

There is no synthetic ordering column: a structured (ROW-typed) value is flattened into one + * table column per field, while a scalar value gets a single value column named after the state. + */ +@Internal +public class OperatorStateDataStreamScanProvider + extends AbstractNonKeyedDataStreamScanProvider { + + public OperatorStateDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier mappingSupplier, + final RowType rowType) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream readState( + SavepointReader savepointReader, OperatorStateTableMapping mapping) throws Exception { + LogicalType valueLogicalType = mapping.getValueLogicalType(); + TypeSerializer valueSerializer = + (TypeSerializer) mapping.getValueTypeSerializer(); + TypeInformation valueTypeInfo = externalTypeInfo(valueLogicalType, valueSerializer); + + DataStream raw; + switch (mapping.getKind()) { + case LIST: + raw = + savepointReader.readListState( + operatorIdentifier, + mapping.getStateName(), + valueTypeInfo, + valueSerializer); + break; + case UNION: + raw = + savepointReader.readUnionState( + operatorIdentifier, + mapping.getStateName(), + valueTypeInfo, + valueSerializer); + break; + default: + throw new UnsupportedOperationException( + "Unsupported operator state kind: " + mapping.getKind()); + } + + return raw.map(new OperatorStateRowMapper(valueLogicalType)) + .returns(InternalTypeInfo.of(rowType)); + } + + /** + * Converts a raw state element into its {@link RowData} representation: a ROW-typed value is + * returned as-is (its fields are the table's flattened columns), while any other value is + * wrapped into a single-column row. + */ + private static class OperatorStateRowMapper implements MapFunction { + + private final LogicalType valueLogicalType; + private final StateValueConverter converter = new StateValueConverter(); + + private OperatorStateRowMapper(LogicalType valueLogicalType) { + this.valueLogicalType = valueLogicalType; + } + + @Override + public RowData map(Object value) { + Object converted = converter.getValue(valueLogicalType, value); + if (valueLogicalType.is(LogicalTypeRoot.ROW)) { + return converted != null + ? (RowData) converted + : new GenericRowData( + RowKind.INSERT, ((RowType) valueLogicalType).getFieldCount()); + } + GenericRowData row = new GenericRowData(RowKind.INSERT, 1); + row.setField(0, converted); + return row; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java new file mode 100644 index 00000000000000..d063e01c6326d3 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; +import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import java.io.Serializable; + +/** + * Maps a {@code ListState}/{@code UnionState} table's schema, in which the state's value is + * flattened directly into the table's columns: one column per field for a structured (ROW-typed) + * value, or a single value column for a scalar one. There is no synthetic ordering column. + * + *

Unlike keyed LIST/MAP state (see {@link FlattenedStateTableMapping}), the value serializer is + * stored flat/unwrapped in the savepoint metadata — there is no wrapping {@code ListSerializer} to + * unwrap. + */ +@Internal +public class OperatorStateTableMapping implements Serializable { + + private static final long serialVersionUID = 1L; + + private final String stateName; + private final SavepointConnectorOptions.StateReaderMode kind; + private final LogicalType valueLogicalType; + private final TypeSerializer valueTypeSerializer; + + public OperatorStateTableMapping( + String stateName, + SavepointConnectorOptions.StateReaderMode kind, + LogicalType valueLogicalType, + TypeSerializer valueTypeSerializer) { + Preconditions.checkArgument( + kind == SavepointConnectorOptions.StateReaderMode.LIST + || kind == SavepointConnectorOptions.StateReaderMode.UNION, + "Operator state tables only support LIST and UNION states, got: " + kind); + this.stateName = stateName; + this.kind = kind; + this.valueLogicalType = valueLogicalType; + this.valueTypeSerializer = valueTypeSerializer; + } + + public String getStateName() { + return stateName; + } + + public SavepointConnectorOptions.StateReaderMode getKind() { + return kind; + } + + public LogicalType getValueLogicalType() { + return valueLogicalType; + } + + public TypeSerializer getValueTypeSerializer() { + return valueTypeSerializer; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema has at least one physical column, matching the flattened (one + * column per value field, or a single value column for a scalar value) LIST/UNION table shape. + * This is a purely structural check; it performs no I/O or class loading. + */ + public static void validateSchema(ResolvedCatalogTable catalogTable) { + if (catalogTable.getResolvedSchema().getColumns().isEmpty()) { + throw new ValidationException( + "LIST/UNION state tables must have at least 1 column, but found none."); + } + } + + /** + * Builds a complete {@link OperatorStateTableMapping}, loading non-keyed operator state + * metadata from the savepoint and resolving the value's logical type and serializer from it. + * + *

Assumes {@link #validateSchema} has already been called. This performs I/O; callers should + * invoke it lazily, deferred to scan time, to keep planning free of savepoint access. + */ + public static OperatorStateTableMapping from( + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig, + SavepointConnectorOptions.StateReaderMode kind) { + + NonKeyedOperatorStateMetadata operatorMetadata = + TableMappingSupport.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + + StateMetaInfoSnapshot stateMetaInfo = + operatorMetadata.operatorStateSnapshots.get(stateName); + if (stateMetaInfo == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' not found in savepoint metadata for operator '" + + operatorIdentifier + + "'."); + } + LogicalType valueLogicalType = + SerializerSnapshotToLogicalTypeConverter.convert( + stateMetaInfo.getTypeSerializerSnapshot( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)); + + SavepointTypeInfoResolver typeResolver = + new SavepointTypeInfoResolver( + operatorMetadata.operatorStateSnapshots, serializerConfig, null); + + TypeSerializer valueTypeSerializer = + typeResolver.resolveFlatValueSerializer( + new RowType.RowField(stateName, valueLogicalType)); + + return new OperatorStateTableMapping( + stateName, kind, valueLogicalType, valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java index 7eafa5ef04e6fd..63240503022588 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java @@ -42,7 +42,10 @@ import static org.apache.flink.state.table.SavepointConnectorOptionsUtil.getOperatorIdentifier; import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR; -/** Dynamic source factory for {@link SavepointDynamicTableSource}. */ +/** + * Dynamic source factory for {@link SavepointDynamicTableSource} and {@link + * NonKeyedDynamicTableSource}. + */ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFactory { @Override @@ -89,6 +92,24 @@ public DynamicTableSource createDynamicTableSource(Context context) { stateBackendType, statePath, operatorIdentifier); + case LIST: + case UNION: + return createOperatorStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier, + readerMode); + case BROADCAST: + return createBroadcastStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); default: throw new IllegalArgumentException("Unsupported state reader mode: " + readerMode); } @@ -276,6 +297,85 @@ private DynamicTableSource createFlattenedWindowDynamicTableSource( WindowFlattenedSavepointDataStreamScanProvider::new); } + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * ListState}/{@code UnionState}. The state name is resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createOperatorStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier, + SavepointConnectorOptions.StateReaderMode readerMode) { + + OperatorStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier mappingSupplier = + () -> + OperatorStateTableMapping.from( + stateName, + statePath, + operatorIdentifier, + serializerConfig, + readerMode); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Operator State Savepoint Table Source", + OperatorStateDataStreamScanProvider::new); + } + + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * BroadcastState}. The state name is resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createBroadcastStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + BroadcastStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier mappingSupplier = + () -> + BroadcastStateTableMapping.from( + context.getCatalogTable(), + stateName, + statePath, + operatorIdentifier, + serializerConfig); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Broadcast State Savepoint Table Source", + BroadcastStateDataStreamScanProvider::new); + } + /** * Validates {@code options} against the required/optional option sets extended with {@link * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the resolved state name — shared @@ -336,13 +436,12 @@ public Set> optionalOptions() { // Multiple values can be read so registering placeholders options.add(STATE_NAME_PLACEHOLDER); - // Selects between the general and flattened keyed-state table schemas; set automatically - // by StateCatalog. + // Selects the table schema / row shape; set automatically by StateCatalog. options.add(STATE_READER_MODE); - // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT (enforced in - // validateAndGetFlattenedStateName); listed here as optional so that generic option - // introspection (docs, Table API tooling) can discover it regardless of mode. + // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT/LIST/UNION/BROADCAST + // (enforced in validateAndGetFlattenedStateName); listed here as optional so that generic + // option introspection (docs, Table API tooling) can discover it regardless of mode. options.add(SavepointConnectorOptions.FLATTENED_STATE_NAME); return options; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java index 52272e698ea894..37e83bcbcca3bf 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java @@ -27,6 +27,7 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.state.api.OperatorIdentifier; import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; import org.apache.flink.state.api.runtime.SavepointLoader.OperatorStateMetadata; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.types.DataType; @@ -151,6 +152,20 @@ static SavepointTypeInfoResolver createTypeResolver( operatorMetadata.keySerializerSnapshot); } + /** + * Preloads the non-keyed (list/union/broadcast) state serializer snapshots of an operator in a + * single I/O operation. Shared by {@link OperatorStateTableMapping} and {@link + * BroadcastStateTableMapping}. + */ + static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String statePath, OperatorIdentifier operatorIdentifier) { + try { + return SavepointLoader.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + } catch (Exception e) { + throw metadataLoadFailure(statePath, operatorIdentifier, e); + } + } + private static RuntimeException metadataLoadFailure( String statePath, OperatorIdentifier operatorIdentifier, Exception cause) { return new RuntimeException( From bdec13a00f5edd9c660a7db46dcca0f5414ad6be Mon Sep 17 00:00:00 2001 From: Gyula Fora Date: Mon, 20 Jul 2026 09:03:53 +0200 Subject: [PATCH 3/3] [FLINK-40180][state-processor-api] Non-keyed reading integration test Adds StateCatalogNonKeyedITCase covering read access to operator ListState/UnionState/BroadcastState through the StateCatalog. --- ...ddedRocksDBStateCatalogNonKeyedITCase.java | 31 ++ .../HashMapStateCatalogNonKeyedITCase.java | 31 ++ .../schema/StateCatalogNonKeyedITCase.java | 433 ++++++++++++++++++ 3 files changed, 495 insertions(+) create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..12b0c4226c3dba --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link StateCatalogNonKeyedITCase} against the embedded RocksDB state backend. */ +public class EmbeddedRocksDBStateCatalogNonKeyedITCase extends StateCatalogNonKeyedITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..d091d6721c780b --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link StateCatalogNonKeyedITCase} against the heap ({@code hashmap}) state backend. */ +public class HashMapStateCatalogNonKeyedITCase extends StateCatalogNonKeyedITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "hashmap"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..5ed7774d387680 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java @@ -0,0 +1,433 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.api.common.functions.RichMapFunction; +import org.apache.flink.api.common.state.BroadcastState; +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.state.FunctionInitializationContext; +import org.apache.flink.runtime.state.FunctionSnapshotContext; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.StateTableUtils; +import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.state.api.utils.SavepointTestBase; +import org.apache.flink.state.catalog.StateCatalog; +import org.apache.flink.state.catalog.TuplePojoField; +import org.apache.flink.state.table.SavepointConnectorOptions; +import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; +import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Integration tests verifying that the savepoint/checkpoint table connector correctly exposes + * non-keyed (operator) state — {@code ListState}, {@code UnionState}, and {@code BroadcastState} — + * via the {@code _list}/{@code _union}/{@code _broadcast} tables. + * + *

Subclassed per state backend since non-keyed state storage is backend-agnostic but the + * catalog/connector layer should still be guarded against backend-specific regressions. + */ +public abstract class StateCatalogNonKeyedITCase extends SavepointTestBase { + + protected abstract Configuration getConfiguration(); + + private static final String UID = "operator-state-writer"; + private static final String LIST_STATE_NAME = "list-values"; + private static final String UNION_STATE_NAME = "union-values"; + private static final String BROADCAST_STATE_NAME = "broadcast-values"; + + private static final String POJO_UID = "pojo-operator-state-writer"; + private static final String POJO_LIST_STATE_NAME = "pojo-list-values"; + private static final String POJO_UNION_STATE_NAME = "pojo-union-values"; + private static final String POJO_BROADCAST_STATE_NAME = "pojo-broadcast-values"; + + /** A plain POJO (public fields, public no-arg constructor) used as an operator state value. */ + public static class NonKeyedEvent { + public String name; + public long value; + + public NonKeyedEvent() {} + + public NonKeyedEvent(String name, long value) { + this.name = name; + this.value = value; + } + } + + @Test + public void testListUnionAndBroadcastStateExposedAsNonKeyedTables() throws Exception { + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(getConfiguration()); + env.setParallelism(1); + + Integer[] data = new Integer[] {1, 2, 3}; + + env.addSource(createSource(data)) + .map(new OperatorStateWriter()) + .uid(UID) + .sinkTo(new DiscardingSink<>()); + + String savepointPath = takeSavepoint(env); + CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(savepointPath); + OperatorIdentifier opId = OperatorIdentifier.forUid(UID); + + NonKeyedStateSchemaInfo schemaInfo = StateTableUtils.getNonKeyedStateSchema(metadata, opId); + assertEquals(StateReaderMode.LIST, schemaInfo.stateSchemas.get(LIST_STATE_NAME).kind); + NonKeyedStateSchemaInfo.StateEntryInfo unionEntry = + schemaInfo.stateSchemas.get(UNION_STATE_NAME); + assertEquals(StateReaderMode.UNION, unionEntry.kind); + assertTrue( + unionEntry.valueLogicalType instanceof RowType, + "Expected the union state's value type to be a structured (ROW) type"); + NonKeyedStateSchemaInfo.StateEntryInfo broadcastEntry = + schemaInfo.stateSchemas.get(BROADCAST_STATE_NAME); + assertEquals(StateReaderMode.BROADCAST, broadcastEntry.kind); + assertNotNull( + broadcastEntry.mapKeyLogicalType, + "Expected a resolved map key type for the broadcast state"); + + TableEnvironment tableEnv = StateCatalogTestUtils.newTableEnv(); + StateCatalog catalog = StateCatalogTestUtils.registerCatalog(tableEnv, savepointPath); + try { + String dbName = catalog.listDatabases().get(0); + tableEnv.useCatalog("state"); + tableEnv.useDatabase(dbName); + + String listTable = + StateCatalog.OPERATOR_UID_PREFIX + + UID + + "_" + + LIST_STATE_NAME + + StateCatalog.LIST_TABLE_SUFFIX; + String unionTable = + StateCatalog.OPERATOR_UID_PREFIX + + UID + + "_" + + UNION_STATE_NAME + + StateCatalog.UNION_TABLE_SUFFIX; + String broadcastTable = + StateCatalog.OPERATOR_UID_PREFIX + + UID + + "_" + + BROADCAST_STATE_NAME + + StateCatalog.BROADCAST_TABLE_SUFFIX; + + List tables = catalog.listTables(dbName); + assertTrue(tables.contains(listTable)); + assertTrue(tables.contains(unionTable)); + assertTrue(tables.contains(broadcastTable)); + + // Non-keyed state is not stored in a state backend, so these tables must not advertise + // a STATE_BACKEND_TYPE option (unlike the four keyed table kinds). + for (String nonKeyedTable : Arrays.asList(listTable, unionTable, broadcastTable)) { + CatalogTable catalogTable = + (CatalogTable) catalog.getTable(new ObjectPath(dbName, nonKeyedTable)); + assertFalse( + catalogTable + .getOptions() + .containsKey(SavepointConnectorOptions.STATE_BACKEND_TYPE.key()), + "Table '" + + nonKeyedTable + + "' should not have a STATE_BACKEND_TYPE option"); + } + + assertScalarListTable(tableEnv, listTable, LIST_STATE_NAME, Arrays.asList(10, 20, 30)); + assertUnionRowTable( + tableEnv, + unionTable, + Arrays.asList(new TuplePojoField("a", 100L), new TuplePojoField("b", 200L))); + + List broadcastRows = + StateCatalogTestUtils.collect( + tableEnv, "SELECT * FROM `" + broadcastTable + "` ORDER BY map_key"); + assertEquals(2, broadcastRows.size()); + assertEquals(1, broadcastRows.get(0).getField("map_key")); + assertEquals("one", broadcastRows.get(0).getField("map_value")); + assertEquals(2, broadcastRows.get(1).getField("map_key")); + assertEquals("two", broadcastRows.get(1).getField("map_value")); + } finally { + catalog.close(); + } + } + + // Verifies the generic PojoToRowDataDeserializer fallback: reads must succeed even with the + // POJO's class hidden from the classpath (simulated via a custom context classloader), instead + // of throwing ClassNotFoundException. + @Test + public void testListUnionAndBroadcastStateWithMissingPojoClassFallsBackToRowData() + throws Exception { + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(getConfiguration()); + env.setParallelism(1); + + Integer[] data = new Integer[] {1, 2, 3}; + + env.addSource(createSource(data)) + .map(new PojoOperatorStateWriter()) + .uid(POJO_UID) + .sinkTo(new DiscardingSink<>()); + + String savepointPath = takeSavepoint(env); + + ClassLoader original = Thread.currentThread().getContextClassLoader(); + ClassLoader hidingEventClass = + new ClassLoader(original) { + @Override + public Class loadClass(String name) throws ClassNotFoundException { + if (name.equals(NonKeyedEvent.class.getName())) { + throw new ClassNotFoundException(name); + } + return super.loadClass(name); + } + }; + Thread.currentThread().setContextClassLoader(hidingEventClass); + try { + TableEnvironment tableEnv = StateCatalogTestUtils.newTableEnv(); + StateCatalog catalog = StateCatalogTestUtils.registerCatalog(tableEnv, savepointPath); + try { + String dbName = catalog.listDatabases().get(0); + tableEnv.useCatalog("state"); + tableEnv.useDatabase(dbName); + + String listTable = + StateCatalog.OPERATOR_UID_PREFIX + + POJO_UID + + "_" + + POJO_LIST_STATE_NAME + + StateCatalog.LIST_TABLE_SUFFIX; + String unionTable = + StateCatalog.OPERATOR_UID_PREFIX + + POJO_UID + + "_" + + POJO_UNION_STATE_NAME + + StateCatalog.UNION_TABLE_SUFFIX; + String broadcastTable = + StateCatalog.OPERATOR_UID_PREFIX + + POJO_UID + + "_" + + POJO_BROADCAST_STATE_NAME + + StateCatalog.BROADCAST_TABLE_SUFFIX; + + List listRows = + StateCatalogTestUtils.collect( + tableEnv, "SELECT * FROM `" + listTable + "` ORDER BY `value`"); + assertEquals(2, listRows.size()); + assertEquals("shared", listRows.get(0).getField("name")); + assertEquals(10L, listRows.get(0).getField("value")); + assertEquals("shared", listRows.get(1).getField("name")); + assertEquals(20L, listRows.get(1).getField("value")); + + List unionRows = + StateCatalogTestUtils.collect( + tableEnv, "SELECT * FROM `" + unionTable + "` ORDER BY `value`"); + assertEquals(2, unionRows.size()); + assertEquals("shared", unionRows.get(0).getField("name")); + assertEquals(10L, unionRows.get(0).getField("value")); + assertEquals("shared", unionRows.get(1).getField("name")); + assertEquals(20L, unionRows.get(1).getField("value")); + + List broadcastRows = + StateCatalogTestUtils.collect( + tableEnv, + "SELECT * FROM `" + broadcastTable + "` ORDER BY map_key"); + assertEquals(2, broadcastRows.size()); + assertEquals(1, broadcastRows.get(0).getField("map_key")); + Row firstValue = broadcastRows.get(0).getFieldAs("map_value"); + assertEquals("one", firstValue.getField("name")); + assertEquals(100L, firstValue.getField("value")); + assertEquals(2, broadcastRows.get(1).getField("map_key")); + Row secondValue = broadcastRows.get(1).getFieldAs("map_value"); + assertEquals("two", secondValue.getField("name")); + assertEquals(200L, secondValue.getField("value")); + } finally { + catalog.close(); + } + } finally { + Thread.currentThread().setContextClassLoader(original); + } + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + /** + * Asserts a scalar-valued {@code _list} table has one column matching {@code expectedValues}. + */ + private static void assertScalarListTable( + TableEnvironment tableEnv, + String tableName, + String valueColumn, + List expectedValues) + throws Exception { + Table table = tableEnv.sqlQuery("SELECT * FROM `" + tableName + "`"); + assertEquals( + Collections.singletonList(valueColumn), table.getResolvedSchema().getColumnNames()); + + List rows = + StateCatalogTestUtils.collect(tableEnv, "SELECT * FROM `" + tableName + "`"); + assertEquals(expectedValues.size(), rows.size()); + + List values = new ArrayList<>(); + for (Row row : rows) { + values.add(row.getFieldAs(valueColumn)); + } + assertEquals(new HashSet<>(expectedValues), new HashSet<>(values)); + } + + /** + * Asserts a structured (ROW-valued) {@code _union} table is flattened into {@code (name, + * score)} columns with no wrapping value column, matching {@code expectedValues}. + */ + private static void assertUnionRowTable( + TableEnvironment tableEnv, String tableName, List expectedValues) + throws Exception { + Table table = tableEnv.sqlQuery("SELECT * FROM `" + tableName + "`"); + assertEquals(Arrays.asList("name", "score"), table.getResolvedSchema().getColumnNames()); + + List rows = + StateCatalogTestUtils.collect(tableEnv, "SELECT * FROM `" + tableName + "`"); + assertEquals(expectedValues.size(), rows.size()); + + List values = new ArrayList<>(); + for (Row row : rows) { + values.add( + new TuplePojoField( + row.getFieldAs("name"), row.getFieldAs("score"))); + } + assertEquals(new HashSet<>(expectedValues), new HashSet<>(values)); + } + + /** + * Registers a {@code ListState}, {@code UnionState}, and {@code BroadcastState} with fixed test + * data. + */ + private static class OperatorStateWriter extends RichMapFunction + implements CheckpointedFunction { + + private transient ListState listState; + private transient ListState unionState; + private transient BroadcastState broadcastState; + + @Override + public Integer map(Integer value) { + return value; + } + + @Override + public void initializeState(FunctionInitializationContext context) throws Exception { + listState = + context.getOperatorStateStore() + .getListState( + new ListStateDescriptor<>(LIST_STATE_NAME, Integer.class)); + unionState = + context.getOperatorStateStore() + .getUnionListState( + new ListStateDescriptor<>( + UNION_STATE_NAME, TuplePojoField.class)); + broadcastState = + context.getOperatorStateStore() + .getBroadcastState( + new MapStateDescriptor<>( + BROADCAST_STATE_NAME, Integer.class, String.class)); + } + + @Override + public void snapshotState(FunctionSnapshotContext context) throws Exception { + listState.update(Arrays.asList(10, 20, 30)); + unionState.update( + Arrays.asList(new TuplePojoField("a", 100L), new TuplePojoField("b", 200L))); + broadcastState.put(1, "one"); + broadcastState.put(2, "two"); + } + } + + /** + * Like {@link OperatorStateWriter}, but with a POJO value type so classpath-fallback reads can + * be exercised. + */ + private static class PojoOperatorStateWriter extends RichMapFunction + implements CheckpointedFunction { + + private transient ListState listState; + private transient ListState unionState; + private transient BroadcastState broadcastState; + + @Override + public Integer map(Integer value) { + return value; + } + + @Override + public void initializeState(FunctionInitializationContext context) throws Exception { + listState = + context.getOperatorStateStore() + .getListState( + new ListStateDescriptor<>( + POJO_LIST_STATE_NAME, NonKeyedEvent.class)); + unionState = + context.getOperatorStateStore() + .getUnionListState( + new ListStateDescriptor<>( + POJO_UNION_STATE_NAME, NonKeyedEvent.class)); + broadcastState = + context.getOperatorStateStore() + .getBroadcastState( + new MapStateDescriptor<>( + POJO_BROADCAST_STATE_NAME, + Integer.class, + NonKeyedEvent.class)); + } + + @Override + public void snapshotState(FunctionSnapshotContext context) throws Exception { + listState.update( + Arrays.asList( + new NonKeyedEvent("shared", 10L), new NonKeyedEvent("shared", 20L))); + unionState.update( + Arrays.asList( + new NonKeyedEvent("shared", 10L), new NonKeyedEvent("shared", 20L))); + broadcastState.put(1, new NonKeyedEvent("one", 100L)); + broadcastState.put(2, new NonKeyedEvent("two", 200L)); + } + } +}