This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit cd7f54c2ae27d40e502f179156802cf50a3a4046 Author: Gyula Fora <[email protected]> AuthorDate: Mon Jul 20 09:02:31 2026 +0200 [FLINK-40180][state-processor-api] Non-keyed/operator state in catalog and factories Wires the non-keyed (ListState/UnionState/BroadcastState) utilities added previously into the catalog and table factory: - StateCatalog: adds the LIST/UNION/BROADCAST table suffixes and getTable() dispatch, and extends candidateTablesForOperator to enumerate one table per non-keyed state (an operator may register only non-keyed state, with no plain per-key "_keyed" table). - SavepointDynamicTableSourceFactory: adds the LIST/UNION/BROADCAST dispatch and createOperatorStateDynamicTableSource/ createBroadcastStateDynamicTableSource, mirroring the keyed/window table source creation. - Adds the DynamicTableSource/ScanProvider/TableMapping implementations for operator (list/union) and broadcast state: AbstractNonKeyedDynamicTableSource, AbstractNonKeyedDataStreamScanProvider, OperatorStateDynamicTableSource, OperatorStateDataStreamScanProvider, OperatorStateTableMapping, BroadcastStateDynamicTableSource, BroadcastStateDataStreamScanProvider, BroadcastStateTableMapping. --- .../state/api/input/OperatorStateInputFormat.java | 7 + .../flink/state/api/runtime/SavepointLoader.java | 109 +++++++++++--- .../apache/flink/state/catalog/StateCatalog.java | 64 ++++++-- .../AbstractNonKeyedDataStreamScanProvider.java | 106 +++++++++++++ .../BroadcastStateDataStreamScanProvider.java | 109 ++++++++++++++ .../state/table/BroadcastStateTableMapping.java | 166 +++++++++++++++++++++ .../state/table/NonKeyedDynamicTableSource.java | 108 ++++++++++++++ .../table/OperatorStateDataStreamScanProvider.java | 123 +++++++++++++++ .../state/table/OperatorStateTableMapping.java | 145 ++++++++++++++++++ .../table/SavepointDynamicTableSourceFactory.java | 111 +++++++++++++- .../flink/state/table/TableMappingSupport.java | 15 ++ 11 files changed, 1023 insertions(+), 40 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java index 65745a1b16a..c768f233fcf 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java @@ -24,6 +24,7 @@ import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.api.common.io.DefaultInputSplitAssigner; import org.apache.flink.api.common.io.RichInputFormat; import org.apache.flink.api.common.io.statistics.BaseStatistics; +import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.core.io.InputSplitAssigner; @@ -36,6 +37,7 @@ import org.apache.flink.runtime.jobgraph.OperatorInstanceID; import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory; import org.apache.flink.state.api.input.splits.OperatorStateInputSplit; import org.apache.flink.streaming.api.operators.StreamOperatorStateContext; import org.apache.flink.util.CollectionUtil; @@ -176,6 +178,11 @@ abstract class OperatorStateInputFormat<OT> extends RichInputFormat<OT, Operator ExecutionConfig executionConfig = KeyedStateInputFormat.deserialize( serializedExecutionConfig, runtimeContext.getUserCodeClassLoader()); + + // Falls back to RowData/GenericRecord for POJO/Avro state whose class is missing from the + // classpath. Must be set before the backend restores, which eagerly restores serializers. + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + final StreamOperatorStateContext context = new StreamOperatorContextBuilder( runtimeContext, diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java index b3ec1a0ce3c..4328388dda8 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java @@ -31,6 +31,8 @@ import org.apache.flink.runtime.state.IncrementalKeyedStateHandle; import org.apache.flink.runtime.state.KeyGroupsStateHandle; import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.OperatorBackendSerializationProxy; +import org.apache.flink.runtime.state.OperatorStateHandle; import org.apache.flink.runtime.state.StreamStateHandle; import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; @@ -41,6 +43,7 @@ import javax.annotation.Nullable; import java.io.DataInputStream; import java.io.IOException; +import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.stream.Collectors; @@ -73,6 +76,27 @@ public final class SavepointLoader { } } + /** + * Non-keyed (operator) state metadata loaded in a single I/O pass. List/union and broadcast + * states are kept in separate maps since a list/union state may share its name with a + * differently-typed broadcast state. + */ + public static final class NonKeyedOperatorStateMetadata { + + /** Per-state serializer snapshots for {@code ListState}/{@code UnionState}, by name. */ + public final Map<String, StateMetaInfoSnapshot> operatorStateSnapshots; + + /** Per-state serializer snapshots for {@code BroadcastState}, by name. */ + public final Map<String, StateMetaInfoSnapshot> broadcastStateSnapshots; + + NonKeyedOperatorStateMetadata( + Map<String, StateMetaInfoSnapshot> operatorStateSnapshots, + Map<String, StateMetaInfoSnapshot> broadcastStateSnapshots) { + this.operatorStateSnapshots = operatorStateSnapshots; + this.broadcastStateSnapshots = broadcastStateSnapshots; + } + } + /** * Takes the given string (representing a pointer to a checkpoint) and resolves it to a file * status for the checkpoint's metadata file. @@ -122,40 +146,65 @@ public final class SavepointLoader { public static OperatorStateMetadata loadOperatorMetadata( String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { - CheckpointMetadata checkpointMetadata = loadSavepointMetadata(savepointPath); + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); - OperatorState operatorState = - checkpointMetadata.getOperatorStates().stream() - .filter( - state -> - operatorIdentifier - .getOperatorId() - .equals(state.getOperatorID())) + KeyedStateHandle keyedStateHandle = + operatorState.getStates().stream() + .flatMap(s -> s.getManagedKeyedState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "Operator " - + operatorIdentifier - + " not found in savepoint")); + "No keyed state found for operator " + + operatorIdentifier)); - KeyedStateHandle keyedStateHandle = + KeyedBackendSerializationProxy<?> proxy = readSerializationProxy(keyedStateHandle); + return new OperatorStateMetadata( + byStateName(proxy.getStateMetaInfoSnapshots()), proxy.getKeySerializerSnapshot()); + } + + /** + * Loads the per-state serializer snapshots of an operator's non-keyed (operator) state — {@code + * ListState}/{@code UnionState}/{@code BroadcastState} — in a single I/O operation. + */ + public static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + + OperatorState operatorState = findOperatorState(savepointPath, operatorIdentifier); + + OperatorStateHandle operatorStateHandle = operatorState.getStates().stream() - .flatMap(s -> s.getManagedKeyedState().stream()) + .flatMap(s -> s.getManagedOperatorState().stream()) .findFirst() .orElseThrow( () -> new IllegalArgumentException( - "No keyed state found for operator " + "No operator state found for operator " + operatorIdentifier)); - KeyedBackendSerializationProxy<?> proxy = readSerializationProxy(keyedStateHandle); - Map<String, StateMetaInfoSnapshot> stateSnapshots = - proxy.getStateMetaInfoSnapshots().stream() - .collect( - Collectors.toMap( - StateMetaInfoSnapshot::getName, Function.identity())); - return new OperatorStateMetadata(stateSnapshots, proxy.getKeySerializerSnapshot()); + OperatorBackendSerializationProxy proxy = readSerializationProxy(operatorStateHandle); + return new NonKeyedOperatorStateMetadata( + byStateName(proxy.getOperatorStateMetaInfoSnapshots()), + byStateName(proxy.getBroadcastStateMetaInfoSnapshots())); + } + + private static OperatorState findOperatorState( + String savepointPath, OperatorIdentifier operatorIdentifier) throws IOException { + return loadSavepointMetadata(savepointPath).getOperatorStates().stream() + .filter(state -> operatorIdentifier.getOperatorId().equals(state.getOperatorID())) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Operator " + + operatorIdentifier + + " not found in savepoint")); + } + + private static Map<String, StateMetaInfoSnapshot> byStateName( + List<StateMetaInfoSnapshot> snapshots) { + return snapshots.stream() + .collect(Collectors.toMap(StateMetaInfoSnapshot::getName, Function.identity())); } private static KeyedBackendSerializationProxy<?> readSerializationProxy( @@ -187,4 +236,22 @@ public final class SavepointLoader { return proxy; } } + + private static OperatorBackendSerializationProxy readSerializationProxy( + OperatorStateHandle stateHandle) throws IOException { + + // Unlike keyed state, an OperatorStateHandle is itself a StreamStateHandle whose stream + // starts with the metadata header, for every state backend. + try (FSDataInputStream inputStream = stateHandle.openInputStream()) { + DataInputViewStreamWrapper inputView = new DataInputViewStreamWrapper(inputStream); + + OperatorBackendSerializationProxy proxy = + new OperatorBackendSerializationProxy( + Thread.currentThread().getContextClassLoader()); + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + proxy.read(inputView); + + return proxy; + } + } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java index 06fab3674b1..290c1cce35b 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java @@ -24,6 +24,7 @@ import org.apache.flink.state.api.OperatorIdentifier; import org.apache.flink.state.api.StateTableUtils; import org.apache.flink.state.api.runtime.SavepointLoader; import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.NonKeyedStateSchemaInfo; import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode; import org.apache.flink.state.table.SavepointConnectorOptions.StateType; import org.apache.flink.table.api.DataTypes; @@ -76,8 +77,8 @@ import java.util.Set; * <ul> * <li>Catalog: the name given at registration time (e.g. {@code "state"}) * <li>Database: one entry per discovered snapshot (e.g. {@code "app1_savepoint-acce1cedsad"}) - * <li>Table: a single view named {@code "metadata"} per database, backed by the {@code - * savepoint_metadata} function from {@code StateModule} + * <li>Table: one per state exposed by the snapshot, plus a view named {@code "metadata"} per + * database, backed by the {@code savepoint_metadata} function from {@code StateModule} * </ul> * * <p>Database names preserve hyphens from the original directory names. Backtick quoting is @@ -113,6 +114,9 @@ public class StateCatalog extends AbstractCatalog { public static final String FLAT_STATE_TABLE_SUFFIX = "_keyed_flat"; public static final String WINDOW_TABLE_SUFFIX = "_windowed"; public static final String FLAT_WINDOW_TABLE_SUFFIX = "_windowed_flat"; + public static final String LIST_TABLE_SUFFIX = "_list"; + public static final String UNION_TABLE_SUFFIX = "_union"; + public static final String BROADCAST_TABLE_SUFFIX = "_broadcast"; private static final CatalogDatabase EMPTY_DATABASE = new CatalogDatabaseImpl(Collections.emptyMap(), ""); @@ -327,6 +331,29 @@ public class StateCatalog extends AbstractCatalog { snapshotPath.get(), resolved.operatorIdentifier); } + case LIST: + case UNION: + { + NonKeyedStateSchemaInfo schemaInfo = + StateTableUtils.getNonKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getOperatorStateCatalogTable( + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } + case BROADCAST: + { + NonKeyedStateSchemaInfo schemaInfo = + StateTableUtils.getNonKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getBroadcastStateCatalogTable( + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } default: throw new IllegalStateException("Unhandled table kind " + resolved.kind); } @@ -620,10 +647,19 @@ public class StateCatalog extends AbstractCatalog { // Operator table helpers // ------------------------------------------------------------------------- + private static final Map<StateReaderMode, String> TABLE_SUFFIXES = + Map.of( + StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, + StateReaderMode.KEYED_FLAT, FLAT_STATE_TABLE_SUFFIX, + StateReaderMode.WINDOWED, WINDOW_TABLE_SUFFIX, + StateReaderMode.WINDOWED_FLAT, FLAT_WINDOW_TABLE_SUFFIX, + StateReaderMode.LIST, LIST_TABLE_SUFFIX, + StateReaderMode.UNION, UNION_TABLE_SUFFIX, + StateReaderMode.BROADCAST, BROADCAST_TABLE_SUFFIX); + /** * Table name for a {@code kind} of operator state, optionally scoped to one flattened/non-keyed - * state (see {@link #OPERATOR_TABLE_SUFFIX}/{@link #FLAT_STATE_TABLE_SUFFIX}/{@link - * #WINDOW_TABLE_SUFFIX}/{@link #FLAT_WINDOW_TABLE_SUFFIX}). + * state (see {@link #TABLE_SUFFIXES}). * * <p>{@code stateName} must be {@code null} for {@link StateReaderMode#KEYED}/{@link * StateReaderMode#WINDOWED} (the general keyed/namespaced table, one per operator) and non-null @@ -631,13 +667,6 @@ public class StateCatalog extends AbstractCatalog { * — the state name alone disambiguates the table since keyed/non-keyed state names are unique * within an operator). */ - private static final Map<StateReaderMode, String> TABLE_SUFFIXES = - Map.of( - StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, - StateReaderMode.KEYED_FLAT, FLAT_STATE_TABLE_SUFFIX, - StateReaderMode.WINDOWED, WINDOW_TABLE_SUFFIX, - StateReaderMode.WINDOWED_FLAT, FLAT_WINDOW_TABLE_SUFFIX); - static String tableName( OperatorIdentifier opId, StateReaderMode kind, @Nullable String stateName) { String base = @@ -702,8 +731,8 @@ public class StateCatalog extends AbstractCatalog { /** * Enumerates every table that {@code opId} contributes: the general keyed/window table (if any - * plain per-key/namespaced state is registered), plus one flattened table per LIST/MAP keyed or - * window state. + * plain per-key/namespaced state is registered), one flattened table per LIST/MAP keyed or + * window state, and one table per non-keyed (list/union/broadcast) state. * * <p>Shared by {@link #listTables} (which collects names for every candidate) and {@link * #resolveTable} (which matches candidates against a target name), so that adding a new state @@ -740,6 +769,15 @@ public class StateCatalog extends AbstractCatalog { } } + // Non-keyed states each get their own table; the entry's kind already carries the table + // shape (LIST/UNION/BROADCAST), so there is no general per-operator table here. + NonKeyedStateSchemaInfo nonKeyedSchemaInfo = + StateTableUtils.getNonKeyedStateSchema(metadata, opId); + for (Map.Entry<String, NonKeyedStateSchemaInfo.StateEntryInfo> entry : + nonKeyedSchemaInfo.stateSchemas.entrySet()) { + candidates.add(new ResolvedTable(opId, entry.getValue().kind, entry.getKey())); + } + return candidates; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java new file mode 100644 index 00000000000..69d29ba4818 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/AbstractNonKeyedDataStreamScanProvider.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.connector.ProviderContext; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.utils.TypeConversions; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Shared scan-time logic for {@link OperatorStateDataStreamScanProvider} and {@link + * BroadcastStateDataStreamScanProvider}: opens the {@link SavepointReader} against the configured + * state backend and resolves the lazy mapping. Subclasses supply the actual {@link SavepointReader} + * call and its row-mapping logic via {@link #readState}. + * + * <p>Unlike keyed state (see {@link AbstractSavepointDataStreamScanProvider}), operator {@code + * ListState}/{@code UnionState}/{@code BroadcastState} have no key column. + */ +@Internal +abstract class AbstractNonKeyedDataStreamScanProvider<M> implements DataStreamScanProvider { + + @Nullable protected final String stateBackendType; + protected final String statePath; + protected final OperatorIdentifier operatorIdentifier; + private final Supplier<M> mappingSupplier; + protected final RowType rowType; + + protected AbstractNonKeyedDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<M> mappingSupplier, + final RowType rowType) { + this.stateBackendType = stateBackendType; + this.statePath = statePath; + this.operatorIdentifier = operatorIdentifier; + this.mappingSupplier = mappingSupplier; + this.rowType = rowType; + } + + @Override + public boolean isBounded() { + return true; + } + + @Override + public DataStream<RowData> produceDataStream( + ProviderContext providerContext, StreamExecutionEnvironment execEnv) { + try { + SavepointReader savepointReader = + AbstractSavepointDataStreamScanProvider.createSavepointReader( + stateBackendType, statePath, execEnv, getClass().getClassLoader()); + + // Resolve the lazy mapping at scan time (class loading deferred from planning). + return readState(savepointReader, mappingSupplier.get()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** + * Drives the actual {@link SavepointReader} call for this scan and maps the result into {@link + * #rowType}-shaped rows. + */ + protected abstract DataStream<RowData> readState(SavepointReader savepointReader, M mapping) + throws Exception; + + /** + * The {@link TypeInformation} the {@link SavepointReader} needs for a raw state value: its SQL + * logical type paired with the serializer the state was written with. + */ + static TypeInformation<Object> externalTypeInfo( + LogicalType logicalType, TypeSerializer<Object> serializer) { + return ExternalTypeInfo.of(TypeConversions.fromLogicalToDataType(logicalType), serializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java new file mode 100644 index 00000000000..7163ce44e7f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateDataStreamScanProvider.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for an operator {@code BroadcastState} table, exposing one + * row per map entry as {@code (map_key, value)}. + */ +@Internal +public class BroadcastStateDataStreamScanProvider + extends AbstractNonKeyedDataStreamScanProvider<BroadcastStateTableMapping> { + + public BroadcastStateDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<BroadcastStateTableMapping> mappingSupplier, + final RowType rowType) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream<RowData> readState( + SavepointReader savepointReader, BroadcastStateTableMapping mapping) throws Exception { + LogicalType mapKeyLogicalType = + rowType.getFields().get(BroadcastStateTableMapping.MAP_KEY_COLUMN_INDEX).getType(); + LogicalType valueLogicalType = + rowType.getFields().get(BroadcastStateTableMapping.VALUE_COLUMN_INDEX).getType(); + + TypeSerializer<Object> mapKeySerializer = + (TypeSerializer<Object>) mapping.getMapKeyTypeSerializer(); + TypeSerializer<Object> valueSerializer = + (TypeSerializer<Object>) mapping.getValueTypeSerializer(); + + DataStream<Tuple2<Object, Object>> raw = + savepointReader.readBroadcastState( + operatorIdentifier, + mapping.getStateName(), + externalTypeInfo(mapKeyLogicalType, mapKeySerializer), + externalTypeInfo(valueLogicalType, valueSerializer), + mapKeySerializer, + valueSerializer); + + return raw.map(new BroadcastStateRowMapper(mapKeyLogicalType, valueLogicalType)) + .returns(InternalTypeInfo.of(rowType)); + } + + /** Converts a raw {@code (map_key, value)} tuple into its {@link RowData} representation. */ + private static class BroadcastStateRowMapper + implements MapFunction<Tuple2<Object, Object>, RowData> { + + private final LogicalType mapKeyLogicalType; + private final LogicalType valueLogicalType; + private final StateValueConverter converter = new StateValueConverter(); + + private BroadcastStateRowMapper( + LogicalType mapKeyLogicalType, LogicalType valueLogicalType) { + this.mapKeyLogicalType = mapKeyLogicalType; + this.valueLogicalType = valueLogicalType; + } + + @Override + public RowData map(Tuple2<Object, Object> entry) { + GenericRowData row = new GenericRowData(RowKind.INSERT, 2); + row.setField( + BroadcastStateTableMapping.MAP_KEY_COLUMN_INDEX, + converter.getValue(mapKeyLogicalType, entry.f0)); + row.setField( + BroadcastStateTableMapping.VALUE_COLUMN_INDEX, + converter.getValue(valueLogicalType, entry.f1)); + return row; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java new file mode 100644 index 00000000000..d1ee179b4a6 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/BroadcastStateTableMapping.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import java.io.Serializable; +import java.util.List; + +/** + * Maps the fixed 2-column schema of a {@code BroadcastState} table: {@code (map_key <key-type> NOT + * NULL, map_value)}, with a primary key on {@code map_key}. + * + * <p>The second column has a fixed name ({@code map_value}) rather than being named after the state + * itself, to avoid collisions with other (reserved) column names; the true state name is instead + * resolved from {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}. + * + * <p>Both the map key and value serializers are stored flat/unwrapped in the savepoint metadata + * (under {@code KEY_SERIALIZER}/{@code VALUE_SERIALIZER}) — unlike a keyed {@code MapState}, there + * is no wrapping {@code MapSerializer} to unwrap. + */ +@Internal +public class BroadcastStateTableMapping implements Serializable { + + private static final long serialVersionUID = 1L; + + public static final int MAP_KEY_COLUMN_INDEX = 0; + public static final int VALUE_COLUMN_INDEX = 1; + + private final String stateName; + private final TypeSerializer<?> mapKeyTypeSerializer; + private final TypeSerializer<?> valueTypeSerializer; + + public BroadcastStateTableMapping( + String stateName, + TypeSerializer<?> mapKeyTypeSerializer, + TypeSerializer<?> valueTypeSerializer) { + this.stateName = stateName; + this.mapKeyTypeSerializer = mapKeyTypeSerializer; + this.valueTypeSerializer = valueTypeSerializer; + } + + public String getStateName() { + return stateName; + } + + public TypeSerializer<?> getMapKeyTypeSerializer() { + return mapKeyTypeSerializer; + } + + public TypeSerializer<?> getValueTypeSerializer() { + return valueTypeSerializer; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema matches the fixed 2-column {@code (map_key, map_value)} + * layout with a primary key on {@code map_key}. This is a purely structural check; it performs + * no I/O or class loading. + */ + public static void validateSchema(ResolvedCatalogTable catalogTable) { + ResolvedSchema schema = catalogTable.getResolvedSchema(); + List<Column> columns = schema.getColumns(); + if (columns.size() != 2) { + throw new ValidationException( + "BROADCAST state tables must have exactly 2 columns " + + "(map_key, map_value), but found " + + columns.size() + + "."); + } + String mapKeyColumnName = columns.get(MAP_KEY_COLUMN_INDEX).getName(); + if (!"map_key".equals(mapKeyColumnName)) { + throw new ValidationException( + "BROADCAST state tables must name their first column 'map_key', " + + "but found '" + + mapKeyColumnName + + "'."); + } + + String valueColumnName = columns.get(VALUE_COLUMN_INDEX).getName(); + if (!"map_value".equals(valueColumnName)) { + throw new ValidationException( + "BROADCAST state tables must name their second column 'map_value', " + + "but found '" + + valueColumnName + + "'."); + } + + List<String> primaryKeyColumns = + schema.getPrimaryKey().map(UniqueConstraint::getColumns).orElse(List.of()); + if (!primaryKeyColumns.equals(List.of(mapKeyColumnName))) { + throw new ValidationException( + "BROADCAST state tables must declare a primary key on '" + + mapKeyColumnName + + "', but found: " + + (primaryKeyColumns.isEmpty() ? "none" : primaryKeyColumns) + + "."); + } + } + + /** + * Builds a complete {@link BroadcastStateTableMapping}, loading non-keyed operator state + * metadata from the savepoint and resolving the key and value serializers from it. + * + * <p>Assumes {@link #validateSchema} has already been called. This performs I/O; callers should + * invoke it lazily, deferred to scan time, to keep planning free of savepoint access. + */ + public static BroadcastStateTableMapping from( + ResolvedCatalogTable catalogTable, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig) { + + NonKeyedOperatorStateMetadata operatorMetadata = + TableMappingSupport.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + + SavepointTypeInfoResolver typeResolver = + new SavepointTypeInfoResolver( + operatorMetadata.broadcastStateSnapshots, serializerConfig, null); + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + RowType rowType = (RowType) physicalDataType.getLogicalType(); + LogicalType valueLogicalType = rowType.getFields().get(VALUE_COLUMN_INDEX).getType(); + // Synthetic RowField: name == actual state name (so metadata lookup succeeds), since the + // value column itself is named after a fixed literal (map_value), not the state. + RowType.RowField valueRowField = new RowType.RowField(stateName, valueLogicalType); + + TypeSerializer<?> mapKeyTypeSerializer = typeResolver.resolveKeySerializer(valueRowField); + TypeSerializer<?> valueTypeSerializer = + typeResolver.resolveFlatValueSerializer(valueRowField); + + return new BroadcastStateTableMapping(stateName, mapKeyTypeSerializer, valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java new file mode 100644 index 00000000000..149deabb0e0 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/NonKeyedDynamicTableSource.java @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Dynamic table source for a single non-keyed operator state: {@code ListState}/{@code UnionState} + * (value flattened into the table's columns) or {@code BroadcastState} (fixed {@code (map_key, + * map_value)} schema). + * + * <p>Unlike keyed state (see {@link AbstractSavepointDynamicTableSource}), non-keyed state has no + * key column, so neither filter nor projection push-down is supported. The scan is delegated to the + * {@link DataStreamScanProvider} supplied by {@link SavepointDynamicTableSourceFactory}, so a + * single table-source class serves every non-keyed state kind. + */ +@Internal +public class NonKeyedDynamicTableSource<M> implements ScanTableSource { + + /** Builds the {@link DataStreamScanProvider} for a given set of scan-time arguments. */ + interface ScanProviderFactory<M> { + DataStreamScanProvider create( + @Nullable String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier, + Supplier<M> mappingSupplier, + RowType rowType); + } + + @Nullable private final String stateBackendType; + private final String statePath; + private final OperatorIdentifier operatorIdentifier; + private final Supplier<M> mappingSupplier; + private final RowType rowType; + private final String summaryString; + private final ScanProviderFactory<M> scanProviderFactory; + + public NonKeyedDynamicTableSource( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<M> mappingSupplier, + final RowType rowType, + final String summaryString, + final ScanProviderFactory<M> scanProviderFactory) { + this.stateBackendType = stateBackendType; + this.statePath = statePath; + this.operatorIdentifier = operatorIdentifier; + this.mappingSupplier = mappingSupplier; + this.rowType = rowType; + this.summaryString = summaryString; + this.scanProviderFactory = scanProviderFactory; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { + return scanProviderFactory.create( + stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + public DynamicTableSource copy() { + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + summaryString, + scanProviderFactory); + } + + @Override + public String asSummaryString() { + return summaryString; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java new file mode 100644 index 00000000000..3c90b35be1f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateDataStreamScanProvider.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.SavepointReader; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; + +import javax.annotation.Nullable; + +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for an operator {@code ListState}/{@code UnionState} table. + * + * <p>There is no synthetic ordering column: a structured (ROW-typed) value is flattened into one + * table column per field, while a scalar value gets a single value column named after the state. + */ +@Internal +public class OperatorStateDataStreamScanProvider + extends AbstractNonKeyedDataStreamScanProvider<OperatorStateTableMapping> { + + public OperatorStateDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<OperatorStateTableMapping> mappingSupplier, + final RowType rowType) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream<RowData> readState( + SavepointReader savepointReader, OperatorStateTableMapping mapping) throws Exception { + LogicalType valueLogicalType = mapping.getValueLogicalType(); + TypeSerializer<Object> valueSerializer = + (TypeSerializer<Object>) mapping.getValueTypeSerializer(); + TypeInformation<Object> valueTypeInfo = externalTypeInfo(valueLogicalType, valueSerializer); + + DataStream<Object> raw; + switch (mapping.getKind()) { + case LIST: + raw = + savepointReader.readListState( + operatorIdentifier, + mapping.getStateName(), + valueTypeInfo, + valueSerializer); + break; + case UNION: + raw = + savepointReader.readUnionState( + operatorIdentifier, + mapping.getStateName(), + valueTypeInfo, + valueSerializer); + break; + default: + throw new UnsupportedOperationException( + "Unsupported operator state kind: " + mapping.getKind()); + } + + return raw.map(new OperatorStateRowMapper(valueLogicalType)) + .returns(InternalTypeInfo.of(rowType)); + } + + /** + * Converts a raw state element into its {@link RowData} representation: a ROW-typed value is + * returned as-is (its fields are the table's flattened columns), while any other value is + * wrapped into a single-column row. + */ + private static class OperatorStateRowMapper implements MapFunction<Object, RowData> { + + private final LogicalType valueLogicalType; + private final StateValueConverter converter = new StateValueConverter(); + + private OperatorStateRowMapper(LogicalType valueLogicalType) { + this.valueLogicalType = valueLogicalType; + } + + @Override + public RowData map(Object value) { + Object converted = converter.getValue(valueLogicalType, value); + if (valueLogicalType.is(LogicalTypeRoot.ROW)) { + return converted != null + ? (RowData) converted + : new GenericRowData( + RowKind.INSERT, ((RowType) valueLogicalType).getFieldCount()); + } + GenericRowData row = new GenericRowData(RowKind.INSERT, 1); + row.setField(0, converted); + return row; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java new file mode 100644 index 00000000000..d063e01c632 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/OperatorStateTableMapping.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; +import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import java.io.Serializable; + +/** + * Maps a {@code ListState}/{@code UnionState} table's schema, in which the state's value is + * flattened directly into the table's columns: one column per field for a structured (ROW-typed) + * value, or a single value column for a scalar one. There is no synthetic ordering column. + * + * <p>Unlike keyed LIST/MAP state (see {@link FlattenedStateTableMapping}), the value serializer is + * stored flat/unwrapped in the savepoint metadata — there is no wrapping {@code ListSerializer} to + * unwrap. + */ +@Internal +public class OperatorStateTableMapping implements Serializable { + + private static final long serialVersionUID = 1L; + + private final String stateName; + private final SavepointConnectorOptions.StateReaderMode kind; + private final LogicalType valueLogicalType; + private final TypeSerializer<?> valueTypeSerializer; + + public OperatorStateTableMapping( + String stateName, + SavepointConnectorOptions.StateReaderMode kind, + LogicalType valueLogicalType, + TypeSerializer<?> valueTypeSerializer) { + Preconditions.checkArgument( + kind == SavepointConnectorOptions.StateReaderMode.LIST + || kind == SavepointConnectorOptions.StateReaderMode.UNION, + "Operator state tables only support LIST and UNION states, got: " + kind); + this.stateName = stateName; + this.kind = kind; + this.valueLogicalType = valueLogicalType; + this.valueTypeSerializer = valueTypeSerializer; + } + + public String getStateName() { + return stateName; + } + + public SavepointConnectorOptions.StateReaderMode getKind() { + return kind; + } + + public LogicalType getValueLogicalType() { + return valueLogicalType; + } + + public TypeSerializer<?> getValueTypeSerializer() { + return valueTypeSerializer; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema has at least one physical column, matching the flattened (one + * column per value field, or a single value column for a scalar value) LIST/UNION table shape. + * This is a purely structural check; it performs no I/O or class loading. + */ + public static void validateSchema(ResolvedCatalogTable catalogTable) { + if (catalogTable.getResolvedSchema().getColumns().isEmpty()) { + throw new ValidationException( + "LIST/UNION state tables must have at least 1 column, but found none."); + } + } + + /** + * Builds a complete {@link OperatorStateTableMapping}, loading non-keyed operator state + * metadata from the savepoint and resolving the value's logical type and serializer from it. + * + * <p>Assumes {@link #validateSchema} has already been called. This performs I/O; callers should + * invoke it lazily, deferred to scan time, to keep planning free of savepoint access. + */ + public static OperatorStateTableMapping from( + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig, + SavepointConnectorOptions.StateReaderMode kind) { + + NonKeyedOperatorStateMetadata operatorMetadata = + TableMappingSupport.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + + StateMetaInfoSnapshot stateMetaInfo = + operatorMetadata.operatorStateSnapshots.get(stateName); + if (stateMetaInfo == null) { + throw new IllegalArgumentException( + "State '" + + stateName + + "' not found in savepoint metadata for operator '" + + operatorIdentifier + + "'."); + } + LogicalType valueLogicalType = + SerializerSnapshotToLogicalTypeConverter.convert( + stateMetaInfo.getTypeSerializerSnapshot( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER)); + + SavepointTypeInfoResolver typeResolver = + new SavepointTypeInfoResolver( + operatorMetadata.operatorStateSnapshots, serializerConfig, null); + + TypeSerializer<?> valueTypeSerializer = + typeResolver.resolveFlatValueSerializer( + new RowType.RowField(stateName, valueLogicalType)); + + return new OperatorStateTableMapping( + stateName, kind, valueLogicalType, valueTypeSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java index 7eafa5ef04e..63240503022 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java @@ -42,7 +42,10 @@ import static org.apache.flink.state.table.SavepointConnectorOptions.STATE_READE import static org.apache.flink.state.table.SavepointConnectorOptionsUtil.getOperatorIdentifier; import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR; -/** Dynamic source factory for {@link SavepointDynamicTableSource}. */ +/** + * Dynamic source factory for {@link SavepointDynamicTableSource} and {@link + * NonKeyedDynamicTableSource}. + */ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFactory { @Override @@ -89,6 +92,24 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac stateBackendType, statePath, operatorIdentifier); + case LIST: + case UNION: + return createOperatorStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier, + readerMode); + case BROADCAST: + return createBroadcastStateDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); default: throw new IllegalArgumentException("Unsupported state reader mode: " + readerMode); } @@ -276,6 +297,85 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac WindowFlattenedSavepointDataStreamScanProvider::new); } + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * ListState}/{@code UnionState}. The state name is resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createOperatorStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier, + SavepointConnectorOptions.StateReaderMode readerMode) { + + OperatorStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier<OperatorStateTableMapping> mappingSupplier = + () -> + OperatorStateTableMapping.from( + stateName, + statePath, + operatorIdentifier, + serializerConfig, + readerMode); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Operator State Savepoint Table Source", + OperatorStateDataStreamScanProvider::new); + } + + /** + * Creates a {@link NonKeyedDynamicTableSource} for a table exposing a single operator {@code + * BroadcastState}. The state name is resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createBroadcastStateDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + BroadcastStateTableMapping.validateSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier<BroadcastStateTableMapping> mappingSupplier = + () -> + BroadcastStateTableMapping.from( + context.getCatalogTable(), + stateName, + statePath, + operatorIdentifier, + serializerConfig); + + return new NonKeyedDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + mappingSupplier, + rowType, + "Broadcast State Savepoint Table Source", + BroadcastStateDataStreamScanProvider::new); + } + /** * Validates {@code options} against the required/optional option sets extended with {@link * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the resolved state name — shared @@ -336,13 +436,12 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac // Multiple values can be read so registering placeholders options.add(STATE_NAME_PLACEHOLDER); - // Selects between the general and flattened keyed-state table schemas; set automatically - // by StateCatalog. + // Selects the table schema / row shape; set automatically by StateCatalog. options.add(STATE_READER_MODE); - // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT (enforced in - // validateAndGetFlattenedStateName); listed here as optional so that generic option - // introspection (docs, Table API tooling) can discover it regardless of mode. + // Required only for STATE_READER_MODE == KEYED_FLAT/WINDOWED_FLAT/LIST/UNION/BROADCAST + // (enforced in validateAndGetFlattenedStateName); listed here as optional so that generic + // option introspection (docs, Table API tooling) can discover it regardless of mode. options.add(SavepointConnectorOptions.FLATTENED_STATE_NAME); return options; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java index 52272e698ea..37e83bcbcca 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/TableMappingSupport.java @@ -27,6 +27,7 @@ import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.state.api.OperatorIdentifier; import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.state.api.runtime.SavepointLoader.NonKeyedOperatorStateMetadata; import org.apache.flink.state.api.runtime.SavepointLoader.OperatorStateMetadata; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.types.DataType; @@ -151,6 +152,20 @@ final class TableMappingSupport { operatorMetadata.keySerializerSnapshot); } + /** + * Preloads the non-keyed (list/union/broadcast) state serializer snapshots of an operator in a + * single I/O operation. Shared by {@link OperatorStateTableMapping} and {@link + * BroadcastStateTableMapping}. + */ + static NonKeyedOperatorStateMetadata loadNonKeyedOperatorMetadata( + String statePath, OperatorIdentifier operatorIdentifier) { + try { + return SavepointLoader.loadNonKeyedOperatorMetadata(statePath, operatorIdentifier); + } catch (Exception e) { + throw metadataLoadFailure(statePath, operatorIdentifier, e); + } + } + private static RuntimeException metadataLoadFailure( String statePath, OperatorIdentifier operatorIdentifier, Exception cause) { return new RuntimeException(
