You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/09/29 11:08:37 UTC
(flink) branch master updated (4a071899941 -> e238d9fb723)
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 4a071899941 [FLINK-40828][core][table] Make VARIANT toString() and result printing never fail
new d32601ce400 [FLINK-40180][state-processor-api] Non-keyed/operator state related utilities
new cd7f54c2ae2 [FLINK-40180][state-processor-api] Non-keyed/operator state in catalog and factories
new e238d9fb723 [FLINK-40180][state-processor-api] Non-keyed reading integration test
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/state/api/StateTableUtils.java | 311 ++++++++++-----
.../state/api/input/OperatorStateInputFormat.java | 7 +
.../flink/state/api/runtime/SavepointLoader.java | 109 +++++-
.../state/api/schema/NonKeyedStateSchemaInfo.java | 68 ++++
.../state/api/schema/OperatorStateSchemaInfo.java | 66 ++++
.../state/api/schema/StateSchemaExtractor.java | 107 +++--
.../apache/flink/state/catalog/StateCatalog.java | 64 ++-
.../AbstractNonKeyedDataStreamScanProvider.java | 106 +++++
.../BroadcastStateDataStreamScanProvider.java | 109 ++++++
.../state/table/BroadcastStateTableMapping.java | 166 ++++++++
...Source.java => NonKeyedDynamicTableSource.java} | 90 ++---
.../table/OperatorStateDataStreamScanProvider.java | 123 ++++++
.../state/table/OperatorStateTableMapping.java | 145 +++++++
.../table/SavepointDynamicTableSourceFactory.java | 111 +++++-
.../flink/state/table/TableMappingSupport.java | 15 +
...EmbeddedRocksDBStateCatalogNonKeyedITCase.java} | 4 +-
...java => HashMapStateCatalogNonKeyedITCase.java} | 4 +-
.../api/schema/StateCatalogNonKeyedITCase.java | 433 +++++++++++++++++++++
18 files changed, 1803 insertions(+), 235 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
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
copy flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/{AbstractSavepointDynamicTableSource.java => NonKeyedDynamicTableSource.java} (52%)
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
copy flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/{EmbeddedRocksDBStateCatalogWindowITCase.java => EmbeddedRocksDBStateCatalogNonKeyedITCase.java} (85%)
copy flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/{HashMapStateCatalogWindowITCase.java => HashMapStateCatalogNonKeyedITCase.java} (86%)
create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java
(flink) 01/03: [FLINK-40180][state-processor-api] Non-keyed/operator state related utilities
Posted by gy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d32601ce400f91266899e7dbbae6e267c49a1100
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Mon Jul 20 08:48:03 2026 +0200
[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.
---
.../apache/flink/state/api/StateTableUtils.java | 311 ++++++++++++++-------
.../state/api/schema/NonKeyedStateSchemaInfo.java | 68 +++++
.../state/api/schema/OperatorStateSchemaInfo.java | 66 +++++
.../state/api/schema/StateSchemaExtractor.java | 107 +++++--
4 files changed, 421 insertions(+), 131 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java
index 731c60c0fac..9bf12c20fbf 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.StateBackendLoader;
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.ArrayType;
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.Set;
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<OperatorIdentifier> getOperatorIdentifiers(CheckpointMetadata metadata) {
return metadata.getOperatorStates().stream()
- .filter(StateTableUtils::hasNonInternalKeyedState)
+ .filter(op -> hasNonInternalKeyedState(op) || hasNonInternalOperatorState(op))
.map(
op ->
op.getOperatorUid()
@@ -113,15 +112,21 @@ public final class StateTableUtils {
}
}
- /**
- * 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<OperatorStateSchemaInfo> 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<String> getKeyedStates(
CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException {
OperatorState opState = findOperatorState(metadata, operatorId);
@@ -133,16 +138,10 @@ public final class StateTableUtils {
}
/**
- * 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.
- *
- * <p>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 @@ public final class StateTableUtils {
}
/**
- * 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 @@ public final class StateTableUtils {
}
/**
- * Builds a {@link CatalogTable} representing all keyed states of an operator.
- *
- * <p>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}.
- *
- * <p>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 final class StateTableUtils {
/**
* Builds a {@link CatalogTable} exposing a single keyed LIST or MAP state flattened into one
- * row per list element / map entry, rather than one row per key.
- *
- * <p>The resulting table has 3 columns, with a composite primary key on {@code state_key} and
- * the sub-key column (the {@code state_key} value repeats across rows belonging to the same
- * key, but the pair uniquely identifies a row). The third column has a fixed name — not the
- * state's own name, to avoid collisions with other (reserved) column names:
+ * 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:
*
* <ul>
- * <li>LIST: {@code (state_key, list_index, list_value)}, primary key {@code (state_key,
- * list_index)}
- * <li>MAP: {@code (state_key, map_key, map_value)}, primary key {@code (state_key, map_key)}
+ * <li>LIST: {@code (state_key, list_index, list_value)}, PK {@code (state_key, list_index)}
+ * <li>MAP: {@code (state_key, map_key, map_value)}, PK {@code (state_key, map_key)}
* </ul>
- *
- * @param metadata the checkpoint metadata the operator belongs to
- * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema}
- * @param stateName the name of the LIST or MAP state to flatten
- * @param statePath the path to the savepoint / checkpoint
- * @param operatorIdentifier identifies the operator whose state to read
- * @return a {@link CatalogTable} ready for registration
*/
public static CatalogTable getFlattenedStateCatalogTable(
CheckpointMetadata metadata,
@@ -466,16 +438,161 @@ public final class StateTableUtils {
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<OperatorStateSchemaInfo> schemas = StateSchemaExtractor.extractOperatorSchema(opState);
+
+ LinkedHashMap<String, NonKeyedStateSchemaInfo.StateEntryInfo> 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<String, String> 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<String, String> 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 @@ public final class StateTableUtils {
}
/**
- * 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}.
- *
- * <p>{@code readerMode} is always set explicitly rather than relying on {@link
- * SavepointConnectorOptions#STATE_READER_MODE}'s default value, so that every catalog table's
- * options unambiguously reflect its schema.
+ * 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<String, String> buildBaseConnectorOptions(
String statePath,
@@ -525,10 +638,9 @@ public final class StateTableUtils {
}
/**
- * 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<String, String> options,
@@ -543,16 +655,12 @@ public final class StateTableUtils {
}
/**
- * 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}.
- *
- * <p>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<String> detectStateBackendType(OperatorState opState) {
Set<String> detectedTypes = new HashSet<>();
@@ -584,8 +692,7 @@ public final class StateTableUtils {
} 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 @@ public final class StateTableUtils {
}
/**
- * 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).
- *
- * <p>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 @@ public final class StateTableUtils {
}
/**
- * 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.
- *
- * <p>An operator may register states under more than one distinct namespace <em>type</em> 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<StateSchemaInfo> 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 00000000000..8f33bb4ecb0
--- /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.
+ *
+ * <p>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<String, StateEntryInfo> stateSchemas;
+
+ public NonKeyedStateSchemaInfo(LinkedHashMap<String, StateEntryInfo> 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 00000000000..31004151863
--- /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.
+ *
+ * <p>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 c58c7bf7737..6774c0e5d84 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.checkpoint.OperatorSubtaskState;
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.
- *
- * <p>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.
- *
- * <p>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.
*
- * <p>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
+ * <p>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<StateSchemaInfo> extractSchema(OperatorState operatorState)
throws IOException {
@@ -84,7 +78,7 @@ public final class StateSchemaExtractor {
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 final class StateSchemaExtractor {
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<StateSchemaInfo> extractSchema(DataInputView in) throws IOException {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
KeyedBackendSerializationProxy<?> proxy = new KeyedBackendSerializationProxy<>(classLoader);
@@ -138,4 +129,70 @@ public final class StateSchemaExtractor {
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<OperatorStateSchemaInfo> 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<OperatorStateSchemaInfo> extractOperatorSchema(DataInputView in)
+ throws IOException {
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ OperatorBackendSerializationProxy proxy =
+ new OperatorBackendSerializationProxy(classLoader);
+ CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create);
+ proxy.read(in);
+
+ List<OperatorStateSchemaInfo> 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;
+ }
}
(flink) 02/03: [FLINK-40180][state-processor-api] Non-keyed/operator state in catalog and factories
Posted by gy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cd7f54c2ae27d40e502f179156802cf50a3a4046
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Mon Jul 20 09:02:31 2026 +0200
[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.
---
.../state/api/input/OperatorStateInputFormat.java | 7 +
.../flink/state/api/runtime/SavepointLoader.java | 109 +++++++++++---
.../apache/flink/state/catalog/StateCatalog.java | 64 ++++++--
.../AbstractNonKeyedDataStreamScanProvider.java | 106 +++++++++++++
.../BroadcastStateDataStreamScanProvider.java | 109 ++++++++++++++
.../state/table/BroadcastStateTableMapping.java | 166 +++++++++++++++++++++
.../state/table/NonKeyedDynamicTableSource.java | 108 ++++++++++++++
.../table/OperatorStateDataStreamScanProvider.java | 123 +++++++++++++++
.../state/table/OperatorStateTableMapping.java | 145 ++++++++++++++++++
.../table/SavepointDynamicTableSourceFactory.java | 111 +++++++++++++-
.../flink/state/table/TableMappingSupport.java | 15 ++
11 files changed, 1023 insertions(+), 40 deletions(-)
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 65745a1b16a..c768f233fcf 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.functions.RuntimeContext;
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.jobgraph.OperatorInstanceID;
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 @@ abstract class OperatorStateInputFormat<OT> extends RichInputFormat<OT, Operator
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 b3ec1a0ce3c..4328388dda8 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.IncrementalKeyedStateHandle;
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 javax.annotation.Nullable;
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 final class SavepointLoader {
}
}
+ /**
+ * 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<String, StateMetaInfoSnapshot> operatorStateSnapshots;
+
+ /** Per-state serializer snapshots for {@code BroadcastState}, by name. */
+ public final Map<String, StateMetaInfoSnapshot> broadcastStateSnapshots;
+
+ NonKeyedOperatorStateMetadata(
+ Map<String, StateMetaInfoSnapshot> operatorStateSnapshots,
+ Map<String, StateMetaInfoSnapshot> 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 final class SavepointLoader {
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<String, StateMetaInfoSnapshot> 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<String, StateMetaInfoSnapshot> byStateName(
+ List<StateMetaInfoSnapshot> snapshots) {
+ return snapshots.stream()
+ .collect(Collectors.toMap(StateMetaInfoSnapshot::getName, Function.identity()));
}
private static KeyedBackendSerializationProxy<?> readSerializationProxy(
@@ -187,4 +236,22 @@ public final class SavepointLoader {
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 06fab3674b1..290c1cce35b 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.OperatorIdentifier;
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 @@ import java.util.Set;
* <ul>
* <li>Catalog: the name given at registration time (e.g. {@code "state"})
* <li>Database: one entry per discovered snapshot (e.g. {@code "app1_savepoint-acce1cedsad"})
- * <li>Table: a single view named {@code "metadata"} per database, backed by the {@code
- * savepoint_metadata} function from {@code StateModule}
+ * <li>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}
* </ul>
*
* <p>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 class StateCatalog extends AbstractCatalog {
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 @@ public class StateCatalog extends AbstractCatalog {
// Operator table helpers
// -------------------------------------------------------------------------
+ private static final Map<StateReaderMode, String> 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}).
*
* <p>{@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 @@ public class StateCatalog extends AbstractCatalog {
* — the state name alone disambiguates the table since keyed/non-keyed state names are unique
* within an operator).
*/
- private static final Map<StateReaderMode, String> 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 @@ public class StateCatalog extends AbstractCatalog {
/**
* 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.
*
* <p>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 @@ public class StateCatalog extends AbstractCatalog {
}
}
+ // 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<String, NonKeyedStateSchemaInfo.StateEntryInfo> 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 00000000000..69d29ba4818
--- /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}.
+ *
+ * <p>Unlike keyed state (see {@link AbstractSavepointDataStreamScanProvider}), operator {@code
+ * ListState}/{@code UnionState}/{@code BroadcastState} have no key column.
+ */
+@Internal
+abstract class AbstractNonKeyedDataStreamScanProvider<M> implements DataStreamScanProvider {
+
+ @Nullable protected final String stateBackendType;
+ protected final String statePath;
+ protected final OperatorIdentifier operatorIdentifier;
+ private final Supplier<M> mappingSupplier;
+ protected final RowType rowType;
+
+ protected AbstractNonKeyedDataStreamScanProvider(
+ @Nullable final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final Supplier<M> 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<RowData> 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<RowData> 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<Object> externalTypeInfo(
+ LogicalType logicalType, TypeSerializer<Object> 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 00000000000..7163ce44e7f
--- /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<BroadcastStateTableMapping> {
+
+ public BroadcastStateDataStreamScanProvider(
+ @Nullable final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final Supplier<BroadcastStateTableMapping> mappingSupplier,
+ final RowType rowType) {
+ super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected DataStream<RowData> 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<Object> mapKeySerializer =
+ (TypeSerializer<Object>) mapping.getMapKeyTypeSerializer();
+ TypeSerializer<Object> valueSerializer =
+ (TypeSerializer<Object>) mapping.getValueTypeSerializer();
+
+ DataStream<Tuple2<Object, Object>> 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<Tuple2<Object, Object>, 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<Object, Object> 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 00000000000..d1ee179b4a6
--- /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 <key-type> NOT
+ * NULL, map_value)}, with a primary key on {@code map_key}.
+ *
+ * <p>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}.
+ *
+ * <p>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<Column> 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<String> 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.
+ *
+ * <p>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 00000000000..149deabb0e0
--- /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).
+ *
+ * <p>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<M> implements ScanTableSource {
+
+ /** Builds the {@link DataStreamScanProvider} for a given set of scan-time arguments. */
+ interface ScanProviderFactory<M> {
+ DataStreamScanProvider create(
+ @Nullable String stateBackendType,
+ String statePath,
+ OperatorIdentifier operatorIdentifier,
+ Supplier<M> mappingSupplier,
+ RowType rowType);
+ }
+
+ @Nullable private final String stateBackendType;
+ private final String statePath;
+ private final OperatorIdentifier operatorIdentifier;
+ private final Supplier<M> mappingSupplier;
+ private final RowType rowType;
+ private final String summaryString;
+ private final ScanProviderFactory<M> scanProviderFactory;
+
+ public NonKeyedDynamicTableSource(
+ @Nullable final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final Supplier<M> mappingSupplier,
+ final RowType rowType,
+ final String summaryString,
+ final ScanProviderFactory<M> 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 00000000000..3c90b35be1f
--- /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.
+ *
+ * <p>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<OperatorStateTableMapping> {
+
+ public OperatorStateDataStreamScanProvider(
+ @Nullable final String stateBackendType,
+ final String statePath,
+ final OperatorIdentifier operatorIdentifier,
+ final Supplier<OperatorStateTableMapping> mappingSupplier,
+ final RowType rowType) {
+ super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected DataStream<RowData> readState(
+ SavepointReader savepointReader, OperatorStateTableMapping mapping) throws Exception {
+ LogicalType valueLogicalType = mapping.getValueLogicalType();
+ TypeSerializer<Object> valueSerializer =
+ (TypeSerializer<Object>) mapping.getValueTypeSerializer();
+ TypeInformation<Object> valueTypeInfo = externalTypeInfo(valueLogicalType, valueSerializer);
+
+ DataStream<Object> 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<Object, RowData> {
+
+ 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 00000000000..d063e01c632
--- /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.
+ *
+ * <p>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.
+ *
+ * <p>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 7eafa5ef04e..63240503022 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.SavepointConnectorOptions.STATE_READE
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 class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac
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 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac
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<OperatorStateTableMapping> 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<BroadcastStateTableMapping> 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 class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac
// 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 52272e698ea..37e83bcbcca 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.ConfigOptions;
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 @@ final class TableMappingSupport {
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(
(flink) 03/03: [FLINK-40180][state-processor-api] Non-keyed reading integration test
Posted by gy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e238d9fb72301e1540c303de4257609059357f8e
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Mon Jul 20 09:03:53 2026 +0200
[FLINK-40180][state-processor-api] Non-keyed reading integration test
Adds StateCatalogNonKeyedITCase covering read access to operator
ListState/UnionState/BroadcastState through the StateCatalog.
---
.../EmbeddedRocksDBStateCatalogNonKeyedITCase.java | 31 ++
.../schema/HashMapStateCatalogNonKeyedITCase.java | 31 ++
.../api/schema/StateCatalogNonKeyedITCase.java | 433 +++++++++++++++++++++
3 files changed, 495 insertions(+)
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 00000000000..12b0c4226c3
--- /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 00000000000..d091d6721c7
--- /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 00000000000..5ed7774d387
--- /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.
+ *
+ * <p>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<String> 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<Row> 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<Row> 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<Row> 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<Row> 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<Integer> expectedValues)
+ throws Exception {
+ Table table = tableEnv.sqlQuery("SELECT * FROM `" + tableName + "`");
+ assertEquals(
+ Collections.singletonList(valueColumn), table.getResolvedSchema().getColumnNames());
+
+ List<Row> rows =
+ StateCatalogTestUtils.collect(tableEnv, "SELECT * FROM `" + tableName + "`");
+ assertEquals(expectedValues.size(), rows.size());
+
+ List<Integer> 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<TuplePojoField> expectedValues)
+ throws Exception {
+ Table table = tableEnv.sqlQuery("SELECT * FROM `" + tableName + "`");
+ assertEquals(Arrays.asList("name", "score"), table.getResolvedSchema().getColumnNames());
+
+ List<Row> rows =
+ StateCatalogTestUtils.collect(tableEnv, "SELECT * FROM `" + tableName + "`");
+ assertEquals(expectedValues.size(), rows.size());
+
+ List<TuplePojoField> values = new ArrayList<>();
+ for (Row row : rows) {
+ values.add(
+ new TuplePojoField(
+ row.<String>getFieldAs("name"), row.<Long>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<Integer, Integer>
+ implements CheckpointedFunction {
+
+ private transient ListState<Integer> listState;
+ private transient ListState<TuplePojoField> unionState;
+ private transient BroadcastState<Integer, String> 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<Integer, Integer>
+ implements CheckpointedFunction {
+
+ private transient ListState<NonKeyedEvent> listState;
+ private transient ListState<NonKeyedEvent> unionState;
+ private transient BroadcastState<Integer, NonKeyedEvent> 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));
+ }
+ }
+}