diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java index 731c60c0fac046..d55efd3ded609c 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java @@ -32,6 +32,8 @@ import org.apache.flink.runtime.state.VoidNamespaceSerializer; import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle; import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.NonKeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.OperatorStateSchemaInfo; import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter; import org.apache.flink.state.api.schema.StateSchemaExtractor; import org.apache.flink.state.api.schema.StateSchemaInfo; @@ -46,6 +48,7 @@ import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter; @@ -67,8 +70,8 @@ import java.util.stream.Collectors; /** - * High-level utility for inspecting and reading keyed state from a checkpoint / savepoint without - * requiring user POJO classes on the classpath. + * High-level utility for inspecting and reading keyed and non-keyed state from a checkpoint / + * savepoint without requiring user POJO classes on the classpath. */ @Internal public final class StateTableUtils { @@ -79,14 +82,14 @@ 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. + * metadata that have at least one non-internal keyed or non-keyed state. * * @param metadata the checkpoint metadata to inspect * @return list of operator identifiers; never null, may be empty */ public static List getOperatorIdentifiers(CheckpointMetadata metadata) { return metadata.getOperatorStates().stream() - .filter(StateTableUtils::hasNonInternalKeyedState) + .filter(op -> hasNonInternalKeyedState(op) || hasNonInternalOperatorState(op)) .map( op -> op.getOperatorUid() @@ -113,12 +116,25 @@ private static boolean hasNonInternalKeyedState(OperatorState op) { } } + private static boolean hasNonInternalOperatorState(OperatorState op) { + try { + List schemas = StateSchemaExtractor.extractOperatorSchema(op); + return schemas.stream().anyMatch(info -> !isInternalState(info.stateName)); + } catch (Exception e) { + LOG.error( + "Could not extract non-keyed state schema for operator '{}': {}. " + + "Excluding from catalog.", + op.getOperatorID(), + e.getMessage()); + return false; + } + } + /** * Returns the names of all keyed states registered by the given operator. * * @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 */ @@ -466,10 +482,187 @@ private static String addFlattenedValueColumns( return subKeyColumnName; } + /** + * Returns the {@link NonKeyedStateSchemaInfo} for the non-keyed (operator) states of the given + * operator — {@code ListState}, {@code UnionState} and {@code BroadcastState} — the ones + * exposed by the {@code _list}/{@code _union}/{@code _broadcast} tables. + * + *

Schema extraction is lenient: POJO field names and types are derived from the serializer + * snapshot and do not require the user POJO class to be on the classpath. States whose schema + * cannot be determined are excluded with a logged error. + * + * @param metadata the checkpoint metadata to inspect + * @param operatorId identifies the operator + * @return schema information covering all registered non-keyed state entries + * @throws IOException if the state header cannot be read + */ + public static NonKeyedStateSchemaInfo getNonKeyedStateSchema( + CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { + OperatorState opState = findOperatorState(metadata, operatorId); + List schemas = StateSchemaExtractor.extractOperatorSchema(opState); + + LinkedHashMap stateSchemas = + new LinkedHashMap<>(); + for (OperatorStateSchemaInfo info : schemas) { + if (isInternalState(info.stateName)) { + continue; + } + try { + LogicalType valueLogicalType = + SerializerSnapshotToLogicalTypeConverter.convert(info.valueSnapshot); + LogicalType mapKeyLogicalType = + info.keySnapshot == null + ? null + : SerializerSnapshotToLogicalTypeConverter.convert( + info.keySnapshot); + stateSchemas.put( + info.stateName, + new NonKeyedStateSchemaInfo.StateEntryInfo( + info.kind, valueLogicalType, mapKeyLogicalType)); + } catch (Exception e) { + logSchemaExtractionFailure("non-keyed ", info.stateName, info.valueSnapshot, e); + } + } + + return new NonKeyedStateSchemaInfo(stateSchemas); + } + + /** + * Builds a {@link CatalogTable} exposing a single {@code ListState} or {@code UnionState} of an + * operator, with one row per list element. + * + *

The state's value is flattened directly into the table's columns instead of being wrapped + * in a single value column: a structured (ROW-typed) value contributes one column per field, + * while a scalar value gets a single column named after the state. There is no synthetic + * ordering column and no primary key, since no column is guaranteed unique across rows. + * + *

Unlike the keyed table builders, this does not add {@link + * SavepointConnectorOptions#STATE_BACKEND_TYPE}: non-keyed state is part of the operator's own + * snapshot rather than of a state backend, so the hint does not apply here. + * + * @param schemaInfo the schema information returned by {@link #getNonKeyedStateSchema} + * @param stateName the name of the LIST or UNION state to expose + * @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 getOperatorStateCatalogTable( + NonKeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + + NonKeyedStateSchemaInfo.StateEntryInfo entryInfo = + findNonKeyedStateEntry(schemaInfo, stateName, operatorIdentifier); + if (entryInfo.kind != SavepointConnectorOptions.StateReaderMode.LIST + && entryInfo.kind != SavepointConnectorOptions.StateReaderMode.UNION) { + throw new IllegalArgumentException( + "Operator state tables are only supported for LIST and UNION states, but '" + + stateName + + "' is " + + entryInfo.kind + + "."); + } + + Schema.Builder schemaBuilder = Schema.newBuilder(); + if (entryInfo.valueLogicalType instanceof RowType) { + for (RowType.RowField field : ((RowType) entryInfo.valueLogicalType).getFields()) { + schemaBuilder.column( + field.getName(), LogicalTypeDataTypeConverter.toDataType(field.getType())); + } + } else { + schemaBuilder.column( + stateName, LogicalTypeDataTypeConverter.toDataType(entryInfo.valueLogicalType)); + } + + Map options = + buildBaseConnectorOptions(statePath, operatorIdentifier, entryInfo.kind); + options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); + + return CatalogTable.newBuilder().schema(schemaBuilder.build()).options(options).build(); + } + + /** + * Builds a {@link CatalogTable} exposing a single {@code BroadcastState} of an operator, with + * one row per broadcast map entry: {@code (map_key NOT NULL, map_value)}, with a + * primary key on {@code map_key}. + * + *

The value column has a fixed name rather than the state's own name, to avoid collisions + * with other (reserved) column names. + * + *

Unlike the keyed table builders, this does not add {@link + * SavepointConnectorOptions#STATE_BACKEND_TYPE}: non-keyed state is part of the operator's own + * snapshot rather than of a state backend, so the hint does not apply here. + * + * @param schemaInfo the schema information returned by {@link #getNonKeyedStateSchema} + * @param stateName the name of the BROADCAST state to expose + * @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 getBroadcastStateCatalogTable( + NonKeyedStateSchemaInfo schemaInfo, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier) { + + NonKeyedStateSchemaInfo.StateEntryInfo entryInfo = + findNonKeyedStateEntry(schemaInfo, stateName, operatorIdentifier); + if (entryInfo.kind != SavepointConnectorOptions.StateReaderMode.BROADCAST) { + throw new IllegalArgumentException( + "Broadcast state tables are only supported for BROADCAST states, but '" + + stateName + + "' is " + + entryInfo.kind + + "."); + } + + Schema schema = + Schema.newBuilder() + .column( + "map_key", + LogicalTypeDataTypeConverter.toDataType(entryInfo.mapKeyLogicalType) + .notNull()) + .column( + "map_value", + LogicalTypeDataTypeConverter.toDataType(entryInfo.valueLogicalType)) + .primaryKeyNamed("PK_map_key", "map_key") + .build(); + + Map options = + buildBaseConnectorOptions( + statePath, + operatorIdentifier, + SavepointConnectorOptions.StateReaderMode.BROADCAST); + options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(), stateName); + + return CatalogTable.newBuilder().schema(schema).options(options).build(); + } + // ------------------------------------------------------------------------- // Private helpers // ------------------------------------------------------------------------- + /** + * 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 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 diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java index 65745a1b16ab55..21cd4a7b3c6dfd 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java @@ -24,6 +24,7 @@ import org.apache.flink.api.common.io.DefaultInputSplitAssigner; import org.apache.flink.api.common.io.RichInputFormat; import org.apache.flink.api.common.io.statistics.BaseStatistics; +import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.core.io.InputSplitAssigner; @@ -36,6 +37,7 @@ import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory; import org.apache.flink.state.api.input.splits.OperatorStateInputSplit; import org.apache.flink.streaming.api.operators.StreamOperatorStateContext; import org.apache.flink.util.CollectionUtil; @@ -176,6 +178,14 @@ public void open(OperatorStateInputSplit split) throws IOException { ExecutionConfig executionConfig = KeyedStateInputFormat.deserialize( serializedExecutionConfig, runtimeContext.getUserCodeClassLoader()); + + // Deserialize any POJO/Avro state whose class is missing from the classpath into + // RowData/GenericRecord instead of failing the restore. Must be registered before the + // backend is restored, since that eagerly restores the previous serializer of every + // registered state. Registering it unconditionally is safe: it is only ever consulted once + // a snapshot has determined that the class it needs is genuinely missing. + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + final StreamOperatorStateContext context = new StreamOperatorContextBuilder( runtimeContext, diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java index b3ec1a0ce3cc19..1e09aa0a1a0aec 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java @@ -31,6 +31,8 @@ import org.apache.flink.runtime.state.KeyGroupsStateHandle; import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.OperatorBackendSerializationProxy; +import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StreamStateHandle; import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; @@ -41,6 +43,7 @@ import java.io.DataInputStream; import java.io.IOException; +import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.stream.Collectors; @@ -73,6 +76,28 @@ public static final class OperatorStateMetadata { } } + /** + * Operator-level metadata for non-keyed (operator) state, loaded in a single I/O pass. The + * list/union states and the broadcast states are kept in two separate maps rather than one + * merged map, since they live in independent state collections and a list/union state may + * legitimately share its name with a differently-typed broadcast state. + */ + public static final class NonKeyedOperatorStateMetadata { + + /** Per-state serializer snapshots for {@code ListState}/{@code UnionState}, by name. */ + public final Map operatorStateSnapshots; + + /** Per-state serializer snapshots for {@code BroadcastState}, by name. */ + public final Map broadcastStateSnapshots; + + NonKeyedOperatorStateMetadata( + Map operatorStateSnapshots, + Map broadcastStateSnapshots) { + this.operatorStateSnapshots = operatorStateSnapshots; + this.broadcastStateSnapshots = broadcastStateSnapshots; + } + } + /** * Takes the given string (representing a pointer to a checkpoint) and resolves it to a file * status for the checkpoint's metadata file. @@ -122,40 +147,70 @@ public static Map loadOperatorStateMetadata( public static OperatorStateMetadata loadOperatorMetadata( String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { - CheckpointMetadata checkpointMetadata = loadSavepointMetadata(savepointPath); + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); - OperatorState operatorState = - checkpointMetadata.getOperatorStates().stream() - .filter( - state -> - operatorIdentifier - .getOperatorId() - .equals(state.getOperatorID())) + KeyedStateHandle keyedStateHandle = + operatorState.getStates().stream() + .flatMap(s -> s.getManagedKeyedState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "Operator " - + operatorIdentifier - + " not found in savepoint")); + "No keyed state found for operator " + + operatorIdentifier)); - KeyedStateHandle keyedStateHandle = + KeyedBackendSerializationProxy proxy = readSerializationProxy(keyedStateHandle); + return new OperatorStateMetadata( + byStateName(proxy.getStateMetaInfoSnapshots()), proxy.getKeySerializerSnapshot()); + } + + /** + * Loads the per-state serializer snapshots of an operator's non-keyed (operator) state — {@code + * ListState}/{@code UnionState}/{@code BroadcastState} — in a single I/O operation. + * + * @param savepointPath Path to the savepoint directory + * @param operatorIdentifier Operator UID or hash + * @return combined non-keyed operator metadata + * @throws IOException If reading fails + */ + public static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); + + OperatorStateHandle operatorStateHandle = operatorState.getStates().stream() - .flatMap(s -> s.getManagedKeyedState().stream()) + .flatMap(s -> s.getManagedOperatorState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "No keyed state found for operator " + "No operator state found for operator " + operatorIdentifier)); - KeyedBackendSerializationProxy proxy = readSerializationProxy(keyedStateHandle); - Map stateSnapshots = - proxy.getStateMetaInfoSnapshots().stream() - .collect( - Collectors.toMap( - StateMetaInfoSnapshot::getName, Function.identity())); - return new OperatorStateMetadata(stateSnapshots, proxy.getKeySerializerSnapshot()); + OperatorBackendSerializationProxy proxy = readSerializationProxy(operatorStateHandle); + return new NonKeyedOperatorStateMetadata( + byStateName(proxy.getOperatorStateMetaInfoSnapshots()), + byStateName(proxy.getBroadcastStateMetaInfoSnapshots())); + } + + private static OperatorState findOperatorState( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + return loadSavepointMetadata(savepointPath).getOperatorStates().stream() + .filter(state -> operatorIdentifier.getOperatorId().equals(state.getOperatorID())) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Operator " + + operatorIdentifier + + " not found in savepoint")); + } + + private static Map byStateName( + List snapshots) { + return snapshots.stream() + .collect(Collectors.toMap(StateMetaInfoSnapshot::getName, Function.identity())); } private static KeyedBackendSerializationProxy readSerializationProxy( @@ -187,4 +242,22 @@ private static KeyedBackendSerializationProxy readSerializationProxy( return proxy; } } + + private static OperatorBackendSerializationProxy readSerializationProxy( + OperatorStateHandle stateHandle) throws IOException { + + // Unlike keyed state, an OperatorStateHandle is itself a StreamStateHandle whose stream + // always 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/api/schema/NonKeyedStateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java new file mode 100644 index 00000000000000..39923ca3015fbf --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/NonKeyedStateSchemaInfo.java @@ -0,0 +1,81 @@ +/* + * 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 + * user POJO classes on the classpath. + * + *

Unlike {@link KeyedStateSchemaInfo}, there is no shared key type across entries: operator + * state has no per-key partitioning, and only {@link StateReaderMode#BROADCAST} entries carry a + * (per-entry) map key type. + */ +@Internal +public final class NonKeyedStateSchemaInfo { + + /** + * Ordered map of registered state names to their entry information. Ordered by the registration + * order found in the savepoint. + */ + public final LinkedHashMap stateSchemas; + + public NonKeyedStateSchemaInfo(LinkedHashMap stateSchemas) { + this.stateSchemas = stateSchemas; + } + + /** Schema information for one non-keyed state entry. */ + public static final class StateEntryInfo { + + /** + * Which non-keyed table kind this entry maps to: {@code LIST}, {@code UNION}, or {@code + * BROADCAST}. + */ + public final StateReaderMode kind; + + /** + * The SQL column logical type of the state's value (e.g. the list element type, or the + * broadcast map's value type). + */ + public final LogicalType valueLogicalType; + + /** + * The resolved {@link LogicalType} of the broadcast map's key. Non-null only for {@link + * StateReaderMode#BROADCAST}. + */ + @Nullable public final LogicalType mapKeyLogicalType; + + public StateEntryInfo( + StateReaderMode kind, + LogicalType valueLogicalType, + @Nullable LogicalType mapKeyLogicalType) { + this.kind = kind; + this.valueLogicalType = valueLogicalType; + this.mapKeyLogicalType = mapKeyLogicalType; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java new file mode 100644 index 00000000000000..fc25618a623edb --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/OperatorStateSchemaInfo.java @@ -0,0 +1,71 @@ +/* + * 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 + * user POJO classes on the classpath. + * + *

Unlike {@link StateSchemaInfo}, there is no per-key serializer: non-keyed state has no key + * concept. {@link #keySnapshot} is only present for {@link StateReaderMode#BROADCAST}, where it + * represents the broadcast map's key. + */ +@Internal +public final class OperatorStateSchemaInfo { + + /** Name of the state as registered by the operator. */ + public final String stateName; + + /** + * Which non-keyed table kind this state maps to: {@code LIST}, {@code UNION}, or {@code + * BROADCAST}. + */ + public final StateReaderMode kind; + + /** + * Serializer snapshot for the state's value type. For {@link StateReaderMode#BROADCAST} this is + * the map's value type; for {@link StateReaderMode#LIST}/{@link StateReaderMode#UNION} this is + * the list element type. + */ + public final TypeSerializerSnapshot valueSnapshot; + + /** + * Serializer snapshot for the broadcast map's key type. Non-null only for {@link + * StateReaderMode#BROADCAST}. + */ + @Nullable public final TypeSerializerSnapshot keySnapshot; + + public OperatorStateSchemaInfo( + String stateName, + StateReaderMode kind, + TypeSerializerSnapshot valueSnapshot, + @Nullable TypeSerializerSnapshot keySnapshot) { + this.stateName = stateName; + this.kind = kind; + this.valueSnapshot = valueSnapshot; + this.keySnapshot = keySnapshot; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java index c58c7bf77370b6..71eb4cb3e265fd 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,28 @@ 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. + * Utility for extracting {@link StateSchemaInfo} / {@link OperatorStateSchemaInfo} from a savepoint + * without instantiating the full state backend or requiring user POJO classes on the classpath. * *

It reads the {@link KeyedBackendSerializationProxy} header that every heap/RocksDB keyed state - * file starts with. + * file starts with, respectively the {@link OperatorBackendSerializationProxy} header that every + * non-keyed (operator) state file starts with. */ @Internal public final class StateSchemaExtractor { @@ -58,7 +63,7 @@ private StateSchemaExtractor() {} * *

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. + * keyed state to describe (see {@link #extractOperatorSchema(OperatorState)}). * *

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 @@ -84,7 +89,7 @@ public static List extractSchema(OperatorState operatorState) metadataHandle = (StreamStateHandle) handle; } if (metadataHandle != null) { - try (java.io.InputStream stream = metadataHandle.openInputStream()) { + try (InputStream stream = metadataHandle.openInputStream()) { return extractSchema(new DataInputViewStreamWrapper(stream)); } } @@ -138,4 +143,83 @@ static List extractSchema(DataInputView in) throws IOException return result; } + + /** + * Reads non-keyed (operator) state schema information — {@code ListState}, {@code UnionState}, + * {@code BroadcastState} — from the first available operator state handle in the given operator + * state. + * + *

Returns an empty list rather than throwing when no operator state handle is found: most + * operators register no list/union/broadcast state at all. + * + *

Unlike keyed state, an {@link OperatorStateHandle} is itself a {@link StreamStateHandle} + * starting with the metadata header, for every state backend. + * + * @param operatorState the operator state from a loaded savepoint / checkpoint metadata + * @return list of schema info, one entry per registered non-keyed state; never null, may be + * empty + * @throws IOException if the state header cannot be read + */ + public static List extractOperatorSchema(OperatorState operatorState) + throws IOException { + + for (OperatorSubtaskState subtask : operatorState.getSubtaskStates().values()) { + for (OperatorStateHandle handle : subtask.getManagedOperatorState()) { + try (InputStream stream = handle.openInputStream()) { + return extractOperatorSchema(new DataInputViewStreamWrapper(stream)); + } + } + } + return Collections.emptyList(); + } + + /** + * Package-private overload that accepts a {@link DataInputView} directly. Allows unit tests to + * inject pre-built byte arrays without a real filesystem. + */ + static List extractOperatorSchema(DataInputView in) + throws IOException { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + OperatorBackendSerializationProxy proxy = + new OperatorBackendSerializationProxy(classLoader); + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + proxy.read(in); + + List result = new ArrayList<>(); + + for (StateMetaInfoSnapshot meta : proxy.getOperatorStateMetaInfoSnapshots()) { + TypeSerializerSnapshot valueSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER); + if (valueSnapshot == null) { + continue; + } + + // Only union-distributed list state is redistributed as a whole on rescale; every + // other distribution mode behaves like a plain (split) ListState here. + String distributionMode = + meta.getOption(CommonOptionsKeys.OPERATOR_STATE_DISTRIBUTION_MODE); + StateReaderMode kind = + OperatorStateHandle.Mode.UNION.name().equals(distributionMode) + ? StateReaderMode.UNION + : StateReaderMode.LIST; + + result.add(new OperatorStateSchemaInfo(meta.getName(), kind, valueSnapshot, null)); + } + + for (StateMetaInfoSnapshot meta : proxy.getBroadcastStateMetaInfoSnapshots()) { + TypeSerializerSnapshot valueSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER); + TypeSerializerSnapshot keySnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.KEY_SERIALIZER); + if (valueSnapshot == null || keySnapshot == null) { + continue; + } + + result.add( + new OperatorStateSchemaInfo( + meta.getName(), StateReaderMode.BROADCAST, valueSnapshot, keySnapshot)); + } + + return result; + } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java index 06fab3674b18e2..290c1cce35ba21 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java @@ -24,6 +24,7 @@ import org.apache.flink.state.api.StateTableUtils; import org.apache.flink.state.api.runtime.SavepointLoader; import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.NonKeyedStateSchemaInfo; import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; import org.apache.flink.state.table.SavepointConnectorOptions.StateType; import org.apache.flink.table.api.DataTypes; @@ -76,8 +77,8 @@ *

* *

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

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

Shared by {@link #listTables} (which collects names for every candidate) and {@link * #resolveTable} (which matches candidates against a target name), so that adding a new state @@ -740,6 +769,15 @@ private static List candidateTablesForOperator( } } + // Non-keyed states each get their own table; the entry's kind already carries the table + // shape (LIST/UNION/BROADCAST), so there is no general per-operator table here. + NonKeyedStateSchemaInfo nonKeyedSchemaInfo = + StateTableUtils.getNonKeyedStateSchema(metadata, opId); + for (Map.Entry entry : + nonKeyedSchemaInfo.stateSchemas.entrySet()) { + candidates.add(new ResolvedTable(opId, entry.getValue().kind, entry.getKey())); + } + return candidates; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java new file mode 100644 index 00000000000000..9a5077c5cd9d8b --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java @@ -0,0 +1,107 @@ +/* + * 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}: opening the {@link SavepointReader} against the configured + * state backend and resolving the lazy mapping. Subclasses supply the actual {@link + * SavepointReader} call and its row-mapping logic via {@link #readState}. + * + *

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

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

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

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

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

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

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

Assumes {@link #validateSchema} has already been called for this table. This performs I/O + * (savepoint metadata loading); callers should invoke it lazily, deferred to scan time, to keep + * planning free of savepoint access. + * + * @param stateName the name of the LIST/UNION state, resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME} + */ + public static OperatorStateTableMapping from( + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig, + SavepointConnectorOptions.StateReaderMode kind) { + + NonKeyedOperatorStateMetadata operatorMetadata = + TableMappingSupport.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + + StateMetaInfoSnapshot stateMetaInfo = + operatorMetadata.operatorStateSnapshots.get(stateName); + if (stateMetaInfo == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' not found in savepoint metadata for operator '" + + operatorIdentifier + + "'."); + } + LogicalType valueLogicalType = + SerializerSnapshotToLogicalTypeConverter.convert( + stateMetaInfo.getTypeSerializerSnapshot( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)); + + SavepointTypeInfoResolver typeResolver = + new SavepointTypeInfoResolver( + operatorMetadata.operatorStateSnapshots, serializerConfig, null); + + TypeSerializer valueTypeSerializer = + typeResolver.resolveFlatValueSerializer( + new RowType.RowField(stateName, valueLogicalType)); + + return new OperatorStateTableMapping( + stateName, kind, valueLogicalType, valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java index 7eafa5ef04e6fd..a1263cd27de18a 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java @@ -42,7 +42,10 @@ import static org.apache.flink.state.table.SavepointConnectorOptionsUtil.getOperatorIdentifier; import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR; -/** Dynamic source factory for {@link SavepointDynamicTableSource}. */ +/** + * Dynamic source factory for {@link SavepointDynamicTableSource} and {@link + * NonKeyedDynamicTableSource}. + */ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFactory { @Override @@ -89,6 +92,24 @@ public DynamicTableSource createDynamicTableSource(Context context) { stateBackendType, statePath, operatorIdentifier); + case LIST: + case UNION: + return createOperatorStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier, + readerMode); + case BROADCAST: + return createBroadcastStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); default: throw new IllegalArgumentException("Unsupported state reader mode: " + readerMode); } @@ -276,6 +297,89 @@ private DynamicTableSource createFlattenedWindowDynamicTableSource( WindowFlattenedSavepointDataStreamScanProvider::new); } + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * ListState}/{@code UnionState} (selected via {@link + * SavepointConnectorOptions#STATE_READER_MODE} being set to {@link + * SavepointConnectorOptions.StateReaderMode#LIST}/{@link + * SavepointConnectorOptions.StateReaderMode#UNION}). The state name is resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createOperatorStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier, + SavepointConnectorOptions.StateReaderMode readerMode) { + + OperatorStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier mappingSupplier = + () -> + OperatorStateTableMapping.from( + stateName, + statePath, + operatorIdentifier, + serializerConfig, + readerMode); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Operator State Savepoint Table Source", + OperatorStateDataStreamScanProvider::new); + } + + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * BroadcastState} (selected via {@link SavepointConnectorOptions#STATE_READER_MODE} being set + * to {@link SavepointConnectorOptions.StateReaderMode#BROADCAST}). The state name is resolved + * from {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createBroadcastStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + BroadcastStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier mappingSupplier = + () -> + BroadcastStateTableMapping.from( + context.getCatalogTable(), + stateName, + statePath, + operatorIdentifier, + serializerConfig); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Broadcast State Savepoint Table Source", + BroadcastStateDataStreamScanProvider::new); + } + /** * Validates {@code options} against the required/optional option sets extended with {@link * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the resolved state name — shared @@ -336,13 +440,12 @@ public Set> optionalOptions() { // Multiple values can be read so registering placeholders options.add(STATE_NAME_PLACEHOLDER); - // Selects between the general and flattened keyed-state table schemas; set automatically - // by StateCatalog. + // Selects the table schema / row shape; set automatically by StateCatalog. options.add(STATE_READER_MODE); - // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT (enforced in - // validateAndGetFlattenedStateName); listed here as optional so that generic option - // introspection (docs, Table API tooling) can discover it regardless of mode. + // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT/LIST/UNION/BROADCAST + // (enforced in validateAndGetFlattenedStateName); listed here as optional so that generic + // option introspection (docs, Table API tooling) can discover it regardless of mode. options.add(SavepointConnectorOptions.FLATTENED_STATE_NAME); return options; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java index 52272e698ea894..37e83bcbcca3bf 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java @@ -27,6 +27,7 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.state.api.OperatorIdentifier; import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; import org.apache.flink.state.api.runtime.SavepointLoader.OperatorStateMetadata; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.types.DataType; @@ -151,6 +152,20 @@ static SavepointTypeInfoResolver createTypeResolver( operatorMetadata.keySerializerSnapshot); } + /** + * Preloads the non-keyed (list/union/broadcast) state serializer snapshots of an operator in a + * single I/O operation. Shared by {@link OperatorStateTableMapping} and {@link + * BroadcastStateTableMapping}. + */ + static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String statePath, OperatorIdentifier operatorIdentifier) { + try { + return SavepointLoader.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + } catch (Exception e) { + throw metadataLoadFailure(statePath, operatorIdentifier, e); + } + } + private static RuntimeException metadataLoadFailure( String statePath, OperatorIdentifier operatorIdentifier, Exception cause) { return new RuntimeException( diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..12b0c4226c3dba --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/EmbeddedRocksDBStateCatalogNonKeyedITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link StateCatalogNonKeyedITCase} against the embedded RocksDB state backend. */ +public class EmbeddedRocksDBStateCatalogNonKeyedITCase extends StateCatalogNonKeyedITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..d091d6721c780b --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/HashMapStateCatalogNonKeyedITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link StateCatalogNonKeyedITCase} against the heap ({@code hashmap}) state backend. */ +public class HashMapStateCatalogNonKeyedITCase extends StateCatalogNonKeyedITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "hashmap"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java new file mode 100644 index 00000000000000..af99dec80e6b17 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateCatalogNonKeyedITCase.java @@ -0,0 +1,443 @@ +/* + * 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. {@code ListState} exercises the + * scalar single-column flattening (no wrapping value column), {@code UnionState} exercises the + * structured (ROW-typed) flattening (one column per field), and {@code BroadcastState} keeps its + * unflattened {@code (map_key, map_value)} shape. + * + *

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