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 ee7fb9d96c16b20bb06aa22ea8d4f0c429acff6d Author: Gyula Fora <[email protected]> AuthorDate: Mon Jul 20 08:26:14 2026 +0200 [FLINK-40179][state-processor-api] Window table support in catalog and factories Wires namespaced (e.g. window-scoped) keyed state into the StateCatalog and SavepointDynamicTableSourceFactory: StateCatalog#getTable/tableName/ candidateTablesForOperator now handle StateReaderMode.WINDOWED/ WINDOWED_FLAT alongside the existing KEYED/KEYED_FLAT tables, and SavepointDynamicTableSourceFactory gains createWindowDynamicTableSource/ createFlattenedWindowDynamicTableSource dispatch. Adds the supporting implementation: WindowStateTableMapping, WindowFlattenedStateTableMapping, WindowKeyedStateReader, WindowFlattenedKeyedStateReader, WindowSavepointDataStreamScanProvider, WindowFlattenedSavepointDataStreamScanProvider, WindowKeyedStateReaderFunction, WindowKeyedStateReaderOperator, and MultiStateKeyAndNamespaceIterator, plus SavepointReader#readWindowKeyedState for programmatic access to namespaced keyed state. Non-keyed/operator state support (LIST/UNION/BROADCAST) is deferred to a later commit. --- .../apache/flink/state/api/SavepointReader.java | 48 +++++ .../functions/WindowKeyedStateReaderFunction.java | 53 +++++ .../input/MultiStateKeyAndNamespaceIterator.java | 101 +++++++++ .../operator/WindowKeyedStateReaderOperator.java | 91 ++++++++ .../apache/flink/state/catalog/StateCatalog.java | 51 ++++- .../table/SavepointDynamicTableSourceFactory.java | 108 ++++++++++ .../table/WindowFlattenedKeyedStateReader.java | 148 +++++++++++++ ...owFlattenedSavepointDataStreamScanProvider.java | 69 ++++++ .../table/WindowFlattenedStateTableMapping.java | 238 +++++++++++++++++++++ .../flink/state/table/WindowKeyedStateReader.java | 110 ++++++++++ .../WindowSavepointDataStreamScanProvider.java | 77 +++++++ .../flink/state/table/WindowStateTableMapping.java | 235 ++++++++++++++++++++ .../WindowKeyedStateReaderFunctionTest.java | 45 ++++ 13 files changed, 1370 insertions(+), 4 deletions(-) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java index ae4bd20332d..a8d77c9dd62 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java @@ -35,12 +35,14 @@ import org.apache.flink.runtime.state.StateBackend; import org.apache.flink.runtime.state.VoidNamespace; import org.apache.flink.state.api.filter.SavepointKeyFilter; import org.apache.flink.state.api.functions.KeyedStateReaderFunction; +import org.apache.flink.state.api.functions.WindowKeyedStateReaderFunction; import org.apache.flink.state.api.input.BroadcastStateInputFormat; import org.apache.flink.state.api.input.KeyedStateInputFormat; import org.apache.flink.state.api.input.ListStateInputFormat; import org.apache.flink.state.api.input.SourceBuilder; import org.apache.flink.state.api.input.UnionStateInputFormat; import org.apache.flink.state.api.input.operator.KeyedStateReaderOperator; +import org.apache.flink.state.api.input.operator.WindowKeyedStateReaderOperator; import org.apache.flink.state.api.runtime.MutableConfig; import org.apache.flink.state.api.runtime.SavepointLoader; import org.apache.flink.state.api.runtime.metadata.SavepointMetadataV2; @@ -55,6 +57,7 @@ import javax.annotation.Nullable; import java.io.IOException; import java.util.Comparator; +import java.util.List; /** The entry point for reading state from a Flink savepoint. */ @PublicEvolving @@ -444,6 +447,51 @@ public class SavepointReader { return SourceBuilder.fromFormat(env, inputFormat, outTypeInfo); } + /** + * Read state registered under an arbitrary namespace (e.g. a window operator's window-scoped + * state) from an operator in a {@code Savepoint}. + * + * @param identifier The identifier of the operator. + * @param function The {@link WindowKeyedStateReaderFunction} that is called for each (key, + * namespace) pair in state. + * @param keyTypeInfo The type information of the key in state. + * @param namespaceSerializer The serializer of the namespace under which the state is + * registered (e.g. a window serializer). + * @param stateNames The names of the namespaced states to read (key, namespace) pairs from. + * @param outTypeInfo The type information of the output of the transform reader function. + * @param keyFilter Optional filter on the state key. When present, input splits whose key + * groups cannot contain any matching key are skipped, and within each split only matching + * keys are iterated. + * @param <K> The type of the key in state. + * @param <OUT> The output type of the transform function. + * @return A {@code DataStream} of objects read from namespaced keyed state. + * @throws IOException If the savepoint does not contain operator state with the given uid. + */ + @Experimental + public <K, OUT> DataStream<OUT> readWindowKeyedState( + OperatorIdentifier identifier, + WindowKeyedStateReaderFunction<K, OUT> function, + TypeInformation<K> keyTypeInfo, + TypeSerializer<Object> namespaceSerializer, + List<String> stateNames, + TypeInformation<OUT> outTypeInfo, + @Nullable SavepointKeyFilter<K> keyFilter) + throws IOException { + + OperatorState operatorState = metadata.getOperatorState(identifier); + KeyedStateInputFormat<K, Object, OUT> inputFormat = + new KeyedStateInputFormat<>( + operatorState, + stateBackend, + MutableConfig.of(env.getConfiguration()), + new WindowKeyedStateReaderOperator<>( + function, keyTypeInfo, namespaceSerializer, stateNames), + env.getConfig(), + keyFilter); + + return SourceBuilder.fromFormat(env, inputFormat, outTypeInfo); + } + /** * Read window state from an operator in a {@code Savepoint}. This method supports reading from * any type of window. diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunction.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunction.java new file mode 100644 index 00000000000..130180c51c5 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunction.java @@ -0,0 +1,53 @@ +/* + * 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.functions; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.functions.AbstractRichFunction; +import org.apache.flink.api.common.state.State; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.util.Collector; + +/** + * Processes (key, namespace) pairs from a restored operator's namespaced (e.g. window-scoped) + * state, invoking {@link #readKey} once per pair. + * + * <p><b>NOTE:</b> Unlike {@link KeyedStateReaderFunction}, state must NOT be accessed via {@link + * #getRuntimeContext()}: the runtime context is always bound to {@code VoidNamespace}, which is + * wrong for namespace-varying reads. Use {@link Context#getState(StateDescriptor)} instead, which + * is bound to the namespace of the (key, namespace) pair currently being processed. + */ +@Internal +public abstract class WindowKeyedStateReaderFunction<K, OUT> extends AbstractRichFunction { + + private static final long serialVersionUID = 1L; + + public abstract void readKey(K key, Object namespace, Context ctx, Collector<OUT> out) + throws Exception; + + /** + * Context that {@link WindowKeyedStateReaderFunction}s can use to access state scoped to the + * (key, namespace) pair currently being processed. Only valid for the duration of a single + * {@link #readKey} call. + */ + public interface Context { + + <S extends State> S getState(StateDescriptor<S, ?> descriptor) throws Exception; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyAndNamespaceIterator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyAndNamespaceIterator.java new file mode 100644 index 00000000000..1a65f4031b6 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyAndNamespaceIterator.java @@ -0,0 +1,101 @@ +/* + * 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.input; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.core.fs.CloseableRegistry; +import org.apache.flink.runtime.state.KeyedStateBackend; +import org.apache.flink.state.api.input.operator.StateReaderOperator; +import org.apache.flink.util.CloseableIterator; +import org.apache.flink.util.Preconditions; + +import java.io.IOException; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.stream.Stream; + +/** + * Reads all (key, namespace) pairs in a state backend across multiple namespaced (e.g. + * window-scoped) states, as the deduplicated union of the pairs stored per given state name: a + * namespace (e.g. a window) may have data in one state but not another, yet should still surface as + * a single row. + * + * <p>The key-group is always reported as {@link StateReaderOperator#UNKNOWN_KEY_GROUP} since the + * backend's per-namespace lookup API does not expose it. + */ +@Internal +public final class MultiStateKeyAndNamespaceIterator<K> + implements CloseableIterator<Tuple3<K, Object, Integer>> { + + private final Iterator<Tuple3<K, Object, Integer>> iterator; + + private final CloseableRegistry registry; + + public MultiStateKeyAndNamespaceIterator( + List<String> stateNames, KeyedStateBackend<K> backend) { + Preconditions.checkNotNull(stateNames, "The state names must not be null"); + Preconditions.checkNotNull(backend, "The keyed state backend must not be null"); + + registry = new CloseableRegistry(); + Stream<Tuple2<K, Object>> merged = + stateNames.stream() + .map(name -> keysAndNamespaces(backend, name)) + .reduce(Stream::concat) + .orElseGet(Stream::empty); + + iterator = + merged.distinct() + .map( + t -> + Tuple3.<K, Object, Integer>of( + t.f0, t.f1, StateReaderOperator.UNKNOWN_KEY_GROUP)) + .iterator(); + } + + private Stream<Tuple2<K, Object>> keysAndNamespaces(KeyedStateBackend<K> backend, String name) { + Stream<Tuple2<K, Object>> stream = backend.getKeysAndNamespaces(name); + try { + registry.registerCloseable(stream::close); + } catch (IOException e) { + throw new RuntimeException("Failed to read keys from configured StateBackend", e); + } + return stream; + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public Tuple3<K, Object, Integer> next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return iterator.next(); + } + + @Override + public void close() throws Exception { + registry.close(); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/operator/WindowKeyedStateReaderOperator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/operator/WindowKeyedStateReaderOperator.java new file mode 100644 index 00000000000..50dbd3e5665 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/operator/WindowKeyedStateReaderOperator.java @@ -0,0 +1,91 @@ +/* + * 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.input.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.State; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.state.api.functions.WindowKeyedStateReaderFunction; +import org.apache.flink.state.api.input.MultiStateKeyAndNamespaceIterator; +import org.apache.flink.state.api.runtime.SavepointRuntimeContext; +import org.apache.flink.util.CloseableIterator; +import org.apache.flink.util.Collector; +import org.apache.flink.util.Preconditions; + +import java.util.List; + +/** + * A {@link StateReaderOperator} for executing a {@link WindowKeyedStateReaderFunction} against + * state registered under an arbitrary (e.g. window) namespace. + * + * <p>Unlike {@link KeyedStateReaderOperator}, the set of state names to read is known upfront (from + * the calling {@code SavepointStateMapping}), so {@link #getKeysAndNamespaces} does not need to go + * through {@code SavepointRuntimeContext}'s state-descriptor registration tracking: it builds a + * {@link MultiStateKeyAndNamespaceIterator} directly from the configured state names. + */ +@Internal +public class WindowKeyedStateReaderOperator<KEY, OUT> + extends StateReaderOperator<WindowKeyedStateReaderFunction<KEY, OUT>, KEY, Object, OUT> { + + private final List<String> stateNames; + + private transient Context context; + + public WindowKeyedStateReaderOperator( + WindowKeyedStateReaderFunction<KEY, OUT> function, + TypeInformation<KEY> keyType, + TypeSerializer<Object> namespaceSerializer, + List<String> stateNames) { + super(function, keyType, namespaceSerializer); + this.stateNames = + Preconditions.checkNotNull(stateNames, "The state names must not be null"); + } + + @Override + public void open() throws Exception { + super.open(); + context = new Context(); + } + + @Override + public void processElement(KEY key, Object namespace, Collector<OUT> out) throws Exception { + context.namespace = namespace; + function.readKey(key, namespace, context, out); + } + + @Override + public CloseableIterator<Tuple3<KEY, Object, Integer>> getKeysAndNamespaces( + SavepointRuntimeContext ctx) throws Exception { + return new MultiStateKeyAndNamespaceIterator<>(stateNames, getKeyedStateBackend()); + } + + private class Context implements WindowKeyedStateReaderFunction.Context { + + private Object namespace; + + @Override + public <S extends State> S getState(StateDescriptor<S, ?> descriptor) throws Exception { + return getKeyedStateBackend() + .getPartitionedState(namespace, namespaceSerializer, descriptor); + } + } +} 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 66c6b174859..06fab3674b1 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 @@ -111,6 +111,8 @@ public class StateCatalog extends AbstractCatalog { public static final String OPERATOR_ID_PREFIX = "id_"; public static final String OPERATOR_TABLE_SUFFIX = "_keyed"; 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"; private static final CatalogDatabase EMPTY_DATABASE = new CatalogDatabaseImpl(Collections.emptyMap(), ""); @@ -302,6 +304,29 @@ public class StateCatalog extends AbstractCatalog { snapshotPath.get(), resolved.operatorIdentifier); } + case WINDOWED: + { + KeyedStateSchemaInfo schemaInfo = + StateTableUtils.getWindowKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getWindowStateCatalogTable( + metadata, + schemaInfo, + snapshotPath.get(), + resolved.operatorIdentifier); + } + case WINDOWED_FLAT: + { + KeyedStateSchemaInfo schemaInfo = + StateTableUtils.getWindowKeyedStateSchema( + metadata, resolved.operatorIdentifier); + return StateTableUtils.getFlattenedWindowStateCatalogTable( + metadata, + schemaInfo, + resolved.stateName, + snapshotPath.get(), + resolved.operatorIdentifier); + } default: throw new IllegalStateException("Unhandled table kind " + resolved.kind); } @@ -597,7 +622,8 @@ public class StateCatalog extends AbstractCatalog { /** * 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}). + * state (see {@link #OPERATOR_TABLE_SUFFIX}/{@link #FLAT_STATE_TABLE_SUFFIX}/{@link + * #WINDOW_TABLE_SUFFIX}/{@link #FLAT_WINDOW_TABLE_SUFFIX}). * * <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 @@ -608,7 +634,9 @@ public class StateCatalog extends AbstractCatalog { private static final Map<StateReaderMode, String> TABLE_SUFFIXES = Map.of( StateReaderMode.KEYED, OPERATOR_TABLE_SUFFIX, - StateReaderMode.KEYED_FLAT, FLAT_STATE_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) { @@ -673,8 +701,9 @@ public class StateCatalog extends AbstractCatalog { } /** - * Enumerates every table that {@code opId} contributes: the general keyed table (if any plain - * per-key state is registered), plus one flattened table per LIST/MAP keyed state. + * 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. * * <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 @@ -697,6 +726,20 @@ public class StateCatalog extends AbstractCatalog { } } + KeyedStateSchemaInfo windowSchemaInfo = + StateTableUtils.getWindowKeyedStateSchema(metadata, opId); + if (!windowSchemaInfo.stateSchemas.isEmpty()) { + candidates.add(new ResolvedTable(opId, StateReaderMode.WINDOWED)); + for (Map.Entry<String, KeyedStateSchemaInfo.StateEntryInfo> entry : + windowSchemaInfo.stateSchemas.entrySet()) { + StateType stateType = entry.getValue().stateType; + if (stateType == StateType.LIST || stateType == StateType.MAP) { + candidates.add( + new ResolvedTable(opId, StateReaderMode.WINDOWED_FLAT, entry.getKey())); + } + } + } + return candidates; } } 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 b3a9c17d1a7..7eafa5ef04e 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 @@ -73,6 +73,22 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac stateBackendType, statePath, operatorIdentifier); + case WINDOWED: + return createWindowDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); + case WINDOWED_FLAT: + return createFlattenedWindowDynamicTableSource( + context, + options, + serializerConfig, + stateBackendType, + statePath, + operatorIdentifier); default: throw new IllegalArgumentException("Unsupported state reader mode: " + readerMode); } @@ -168,6 +184,98 @@ public class SavepointDynamicTableSourceFactory implements DynamicTableSourceFac FlattenedSavepointDataStreamScanProvider::new); } + /** + * Creates a {@link SavepointDynamicTableSource} for the general namespaced (e.g. window-scoped) + * keyed state table (selected via {@link SavepointConnectorOptions#STATE_READER_MODE} being set + * to {@link SavepointConnectorOptions.StateReaderMode#WINDOWED}). + */ + private DynamicTableSource createWindowDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + Set<ConfigOption<?>> requiredOptions = new HashSet<>(requiredOptions()); + Set<ConfigOption<?>> optionalOptions = new HashSet<>(optionalOptions()); + + // Validate schema and register per-field options eagerly (no class loading) so that + // option validation passes at planning time. + int[] keyAndWindowColumnIndices = + WindowStateTableMapping.validateAndExtractKeyAndWindowColumns( + context.getCatalogTable(), optionalOptions); + int keyColumnIndex = keyAndWindowColumnIndices[0]; + + validateOptions(options, requiredOptions, optionalOptions); + + // Defer I/O and class loading to scan time by creating the mapping lazily. + Supplier<WindowStateTableMapping> mappingSupplier = + () -> + WindowStateTableMapping.from( + context.getCatalogTable(), + options, + statePath, + operatorIdentifier, + serializerConfig); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + return new SavepointDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + keyColumnIndex, + mappingSupplier, + rowType, + "Window Savepoint Table Source", + WindowSavepointDataStreamScanProvider::new); + } + + /** + * Creates a {@link FlattenedSavepointDynamicTableSource} for a table exposing a single + * flattened namespaced (e.g. window-scoped) LIST/MAP state (selected via {@link + * SavepointConnectorOptions#STATE_READER_MODE} being set to {@link + * SavepointConnectorOptions.StateReaderMode#WINDOWED_FLAT}). The state name is resolved from + * {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}. + */ + private DynamicTableSource createFlattenedWindowDynamicTableSource( + Context context, + Configuration options, + SerializerConfig serializerConfig, + String stateBackendType, + String statePath, + OperatorIdentifier operatorIdentifier) { + + SavepointConnectorOptions.StateType stateType = + WindowFlattenedStateTableMapping.validateFlattenedSchema(context.getCatalogTable()); + + RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType(); + + String stateName = validateAndGetFlattenedStateName(options); + + // Defer I/O to scan time by creating the mapping lazily. + Supplier<WindowFlattenedStateTableMapping> mappingSupplier = + () -> + WindowFlattenedStateTableMapping.from( + context.getCatalogTable(), + stateName, + statePath, + operatorIdentifier, + serializerConfig, + stateType); + + return new FlattenedSavepointDynamicTableSource<>( + stateBackendType, + statePath, + operatorIdentifier, + WindowFlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, + mappingSupplier, + rowType, + "Flattened Window Savepoint Table Source", + WindowFlattenedSavepointDataStreamScanProvider::new); + } + /** * Validates {@code options} against the required/optional option sets extended with {@link * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the resolved state name — shared diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedKeyedStateReader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedKeyedStateReader.java new file mode 100644 index 00000000000..5ab58db572f --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedKeyedStateReader.java @@ -0,0 +1,148 @@ +/* + * 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.api.common.state.ListState; +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.State; +import org.apache.flink.state.api.functions.WindowKeyedStateReaderFunction; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.Collector; + +import java.util.Map; + +/** + * Reads a single flattened namespaced (e.g. window-scoped) keyed list/map state, emitting one row + * per list element / map entry per {@code (key, namespace)} pair: {@code (state_key, state_window, + * index, value)} for LIST, {@code (state_key, state_window, map_key, value)} for MAP. + * + * <p>Unlike {@link FlattenedKeyedStateReader}, the state is not cached in {@code open()}: the + * namespace varies per {@code (key, namespace)} pair being processed, so it is fetched fresh, + * scoped to the current namespace, via {@link WindowKeyedStateReaderFunction.Context#getState}. + */ +@SuppressWarnings({"rawtypes", "unchecked"}) +public class WindowFlattenedKeyedStateReader + extends WindowKeyedStateReaderFunction<Object, RowData> { + + private final RowType rowType; + private final WindowFlattenedStateTableMapping mapping; + private final StateValueConverter converter = new StateValueConverter(); + + public WindowFlattenedKeyedStateReader( + final RowType rowType, final WindowFlattenedStateTableMapping mapping) { + this.rowType = rowType; + this.mapping = mapping; + } + + @Override + public void readKey(Object key, Object namespace, Context context, Collector<RowData> out) + throws Exception { + LogicalType keyLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX) + .getType(); + Object convertedKey = converter.getValue(keyLogicalType, key); + + LogicalType windowLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.WINDOW_COLUMN_INDEX) + .getType(); + Object convertedWindow = converter.getValue(windowLogicalType, namespace); + + State state = context.getState(mapping.getStateDescriptor()); + + switch (mapping.getStateType()) { + case LIST: + readList(convertedKey, convertedWindow, (ListState) state, out); + break; + + case MAP: + readMap(convertedKey, convertedWindow, (MapState) state, out); + break; + + default: + throw new UnsupportedOperationException( + "Unsupported flattened state type: " + mapping.getStateType()); + } + } + + private void readList( + Object convertedKey, Object convertedWindow, ListState state, Collector<RowData> out) + throws Exception { + LogicalType valueLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.VALUE_COLUMN_INDEX) + .getType(); + LogicalType indexLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX) + .getType(); + + Iterable<Object> values = (Iterable<Object>) state.get(); + converter.writeListRows( + values, + () -> { + GenericRowData row = new GenericRowData(RowKind.INSERT, 4); + row.setField( + WindowFlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, convertedKey); + row.setField( + WindowFlattenedStateTableMapping.WINDOW_COLUMN_INDEX, convertedWindow); + return row; + }, + WindowFlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX, + WindowFlattenedStateTableMapping.VALUE_COLUMN_INDEX, + indexLogicalType, + valueLogicalType, + out); + } + + private void readMap( + Object convertedKey, Object convertedWindow, MapState state, Collector<RowData> out) + throws Exception { + LogicalType valueLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.VALUE_COLUMN_INDEX) + .getType(); + LogicalType mapKeyLogicalType = + rowType.getFields() + .get(WindowFlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX) + .getType(); + + Iterable<Map.Entry<Object, Object>> entries = state.entries(); + converter.writeMapRows( + entries, + () -> { + GenericRowData row = new GenericRowData(RowKind.INSERT, 4); + row.setField( + WindowFlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX, convertedKey); + row.setField( + WindowFlattenedStateTableMapping.WINDOW_COLUMN_INDEX, convertedWindow); + return row; + }, + WindowFlattenedStateTableMapping.SUB_KEY_COLUMN_INDEX, + WindowFlattenedStateTableMapping.VALUE_COLUMN_INDEX, + mapKeyLogicalType, + valueLogicalType, + out); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedSavepointDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedSavepointDataStreamScanProvider.java new file mode 100644 index 00000000000..95198f4a9cf --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedSavepointDataStreamScanProvider.java @@ -0,0 +1,69 @@ +/* + * 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.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.state.api.filter.SavepointKeyFilter; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.function.Supplier; + +/** + * Savepoint data stream scan provider for a single flattened namespaced (e.g. window-scoped) keyed + * LIST/MAP state, emitting one row per list element / map entry per {@code (key, namespace)} pair + * (see {@link WindowFlattenedKeyedStateReader}). + */ +@SuppressWarnings("rawtypes") +public class WindowFlattenedSavepointDataStreamScanProvider + extends AbstractSingleColumnScanProvider<WindowFlattenedStateTableMapping> { + + public WindowFlattenedSavepointDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<WindowFlattenedStateTableMapping> mappingSupplier, + final RowType rowType, + @Nullable final SavepointKeyFilter keyFilter) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType, keyFilter); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream<RowData> readState( + SavepointReader savepointReader, WindowFlattenedStateTableMapping mapping) + throws Exception { + return savepointReader.readWindowKeyedState( + operatorIdentifier, + new WindowFlattenedKeyedStateReader(rowType, mapping), + (TypeInformation) mapping.getKeyTypeInfo(), + (TypeSerializer) mapping.getNamespaceSerializer(), + List.of(mapping.getStateName()), + InternalTypeInfo.of(rowType), + keyFilter); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedStateTableMapping.java new file mode 100644 index 00000000000..fb2f266f658 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowFlattenedStateTableMapping.java @@ -0,0 +1,238 @@ +/* + * 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.state.StateDescriptor; +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.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.types.DataType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.List; + +/** + * Maps the fixed 4-column schema of a flattened namespaced (e.g. window-scoped) keyed list/map + * state table: + * + * <ul> + * <li>LIST: {@code (state_key, state_window, list_index, list_value)} + * <li>MAP: {@code (state_key, state_window, map_key, map_value)} + * </ul> + * + * <p>The fourth column has a fixed name ({@code list_value}/{@code map_value}) rather than being + * named after the flattened state itself, to avoid collisions with other (reserved) column names; + * the true state name is instead resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME}. + * + * <p>Mirrors {@link FlattenedStateTableMapping}, shifted by one column to make room for {@code + * state_window}. As with {@link WindowStateTableMapping}, no primary key is declared on the table + * (Flink's primary-key validation for arbitrary {@code ROW}-typed columns such as {@code + * state_window} is not reliable), so columns are identified by fixed position, exactly as in {@link + * FlattenedStateTableMapping}. + */ +@Internal +public class WindowFlattenedStateTableMapping implements Serializable, SingleColumnStateMapping { + + private static final long serialVersionUID = 1L; + + public static final int STATE_KEY_COLUMN_INDEX = 0; + public static final int WINDOW_COLUMN_INDEX = 1; + public static final int SUB_KEY_COLUMN_INDEX = 2; + public static final int VALUE_COLUMN_INDEX = 3; + + private final String stateName; + private final SavepointConnectorOptions.StateType stateType; + private final TypeInformation<?> keyTypeInfo; + @Nullable private final TypeSerializer<?> mapKeyTypeSerializer; + private final TypeSerializer<?> valueTypeSerializer; + @Nullable private final TypeSerializer<Object> namespaceSerializer; + @Nullable private StateDescriptor stateDescriptor; + + public WindowFlattenedStateTableMapping( + String stateName, + SavepointConnectorOptions.StateType stateType, + TypeInformation<?> keyTypeInfo, + @Nullable TypeSerializer<?> mapKeyTypeSerializer, + TypeSerializer<?> valueTypeSerializer, + @Nullable TypeSerializer<Object> namespaceSerializer) { + Preconditions.checkArgument( + stateType == SavepointConnectorOptions.StateType.LIST + || stateType == SavepointConnectorOptions.StateType.MAP, + "Flattened state tables only support LIST and MAP states, got: " + stateType); + this.stateName = stateName; + this.stateType = stateType; + this.keyTypeInfo = keyTypeInfo; + this.mapKeyTypeSerializer = mapKeyTypeSerializer; + this.valueTypeSerializer = valueTypeSerializer; + this.namespaceSerializer = namespaceSerializer; + } + + @Override + public String getStateName() { + return stateName; + } + + @Override + public SavepointConnectorOptions.StateType getStateType() { + return stateType; + } + + @Override + public TypeInformation<?> getKeyTypeInfo() { + return keyTypeInfo; + } + + @Override + @Nullable + public TypeSerializer<?> getMapKeyTypeSerializer() { + return mapKeyTypeSerializer; + } + + @Override + public TypeSerializer<?> getValueTypeSerializer() { + return valueTypeSerializer; + } + + @Nullable + public TypeSerializer<Object> getNamespaceSerializer() { + return namespaceSerializer; + } + + @Override + @SuppressWarnings("rawtypes") + public void setStateDescriptor(StateDescriptor stateDescriptor) { + this.stateDescriptor = stateDescriptor; + } + + @Nullable + @SuppressWarnings("rawtypes") + public StateDescriptor getStateDescriptor() { + return stateDescriptor; + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates that the table schema matches the fixed 4-column flattened window layout, and + * returns the state type (LIST or MAP), inferred from whether the third column is named {@code + * list_index} or {@code map_key}. This is a purely structural check; it performs no I/O or + * class loading. + */ + public static SavepointConnectorOptions.StateType validateFlattenedSchema( + ResolvedCatalogTable catalogTable) { + ResolvedSchema schema = catalogTable.getResolvedSchema(); + List<Column> columns = schema.getColumns(); + if (columns.size() != 4) { + throw new ValidationException( + "Flattened namespaced (window) keyed state tables must have exactly 4 columns " + + "(state_key, state_window, list_index/map_key, list_value/map_value), " + + "but found " + + columns.size() + + "."); + } + DataType physicalDataType = schema.toPhysicalRowDataType(); + Preconditions.checkArgument( + physicalDataType.getLogicalType().is(LogicalTypeRoot.ROW), + "Row data type expected."); + + String windowColumnName = columns.get(WINDOW_COLUMN_INDEX).getName(); + if (!windowColumnName.equals(WindowStateTableMapping.WINDOW_COLUMN_NAME)) { + throw new ValidationException( + "Flattened namespaced (window) keyed state tables must name their second " + + "column '" + + WindowStateTableMapping.WINDOW_COLUMN_NAME + + "', but found '" + + windowColumnName + + "'."); + } + + String subKeyColumnName = columns.get(SUB_KEY_COLUMN_INDEX).getName(); + String valueColumnName = columns.get(VALUE_COLUMN_INDEX).getName(); + return TableMappingSupport.inferFlattenedStateTypeAndValidateValueColumn( + "Flattened namespaced (window) keyed state tables", + "third", + "fourth", + subKeyColumnName, + valueColumnName); + } + + /** + * Builds a complete {@link WindowFlattenedStateTableMapping}, loading operator state metadata + * from the savepoint and resolving serializers, key type and namespace serializer from it. + * + * <p>Assumes {@link #validateFlattenedSchema} has already been called for this table. + * + * <p>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 flattened LIST/MAP state, resolved from {@link + * SavepointConnectorOptions#FLATTENED_STATE_NAME} + */ + public static WindowFlattenedStateTableMapping from( + ResolvedCatalogTable catalogTable, + String stateName, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig, + SavepointConnectorOptions.StateType stateType) { + + SavepointTypeInfoResolver typeResolver = + TableMappingSupport.createTypeResolver( + statePath, operatorIdentifier, serializerConfig); + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + RowType rowType = (RowType) physicalDataType.getLogicalType(); + + TableMappingSupport.FlattenedSerializers serializers = + TableMappingSupport.resolveFlattenedSerializers( + rowType, + typeResolver, + stateName, + stateType, + STATE_KEY_COLUMN_INDEX, + SUB_KEY_COLUMN_INDEX, + VALUE_COLUMN_INDEX); + + @SuppressWarnings("unchecked") + TypeSerializer<Object> namespaceSerializer = + (TypeSerializer<Object>) typeResolver.resolveNamespaceSerializer(stateName); + + return new WindowFlattenedStateTableMapping( + stateName, + stateType, + serializers.keyTypeInfo, + serializers.mapKeyTypeSerializer, + serializers.valueTypeSerializer, + namespaceSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowKeyedStateReader.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowKeyedStateReader.java new file mode 100644 index 00000000000..110bdcbe7f1 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowKeyedStateReader.java @@ -0,0 +1,110 @@ +/* + * 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.api.common.state.ListState; +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.State; +import org.apache.flink.state.api.functions.WindowKeyedStateReaderFunction; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.Collector; + +import java.util.List; + +/** + * Namespaced (e.g. window-scoped) keyed state reader function for value, list and map state types. + * + * <p>Unlike {@link KeyedStateReader}, states are not registered/cached in {@code open()}: the + * namespace varies per {@code (key, namespace)} pair being processed, so each state must be fetched + * fresh, scoped to the current namespace, via {@link + * WindowKeyedStateReaderFunction.Context#getState}. + */ +@SuppressWarnings({"rawtypes", "unchecked"}) +public class WindowKeyedStateReader extends WindowKeyedStateReaderFunction<Object, RowData> { + + private final RowType rowType; + private final WindowStateTableMapping mapping; + private final StateValueConverter converter = new StateValueConverter(); + + public WindowKeyedStateReader(final RowType rowType, final WindowStateTableMapping mapping) { + this.rowType = rowType; + this.mapping = mapping; + } + + @Override + public void readKey(Object key, Object namespace, Context context, Collector<RowData> out) + throws Exception { + GenericRowData row = new GenericRowData(RowKind.INSERT, rowType.getFieldCount()); + + List<RowType.RowField> fields = rowType.getFields(); + + int keyColumnIndex = mapping.getKeyColumnIndex(); + if (keyColumnIndex >= 0) { + LogicalType keyLogicalType = fields.get(keyColumnIndex).getType(); + row.setField(keyColumnIndex, converter.getValue(keyLogicalType, key)); + } + + int windowColumnIndex = mapping.getWindowColumnIndex(); + if (windowColumnIndex >= 0) { + LogicalType windowLogicalType = fields.get(windowColumnIndex).getType(); + row.setField(windowColumnIndex, converter.getValue(windowLogicalType, namespace)); + } + + for (StateValueColumnConfiguration columnConfig : mapping.getValueColumns()) { + LogicalType valueLogicalType = fields.get(columnConfig.getColumnIndex()).getType(); + switch (columnConfig.getStateType()) { + case VALUE: + State valueState = context.getState(columnConfig.getStateDescriptor()); + row.setField( + columnConfig.getColumnIndex(), + converter.getValue( + valueLogicalType, + StateValueConverter.readValueLikeState( + valueState, columnConfig.getActualStateKind()))); + break; + + case LIST: + ListState listState = + (ListState) context.getState(columnConfig.getStateDescriptor()); + row.setField( + columnConfig.getColumnIndex(), + converter.getValue(valueLogicalType, listState.get())); + break; + + case MAP: + MapState mapState = + (MapState) context.getState(columnConfig.getStateDescriptor()); + row.setField( + columnConfig.getColumnIndex(), + converter.getValue(valueLogicalType, mapState.entries())); + break; + + default: + throw new UnsupportedOperationException( + "Unsupported state type: " + columnConfig.getStateType()); + } + } + + out.collect(row); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowSavepointDataStreamScanProvider.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowSavepointDataStreamScanProvider.java new file mode 100644 index 00000000000..89b2c8e83e0 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowSavepointDataStreamScanProvider.java @@ -0,0 +1,77 @@ +/* + * 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.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.state.api.filter.SavepointKeyFilter; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +/** + * Savepoint data stream scan provider for the general namespaced (e.g. window-scoped) keyed state + * table, emitting one row per {@code (key, namespace)} pair (see {@link WindowKeyedStateReader}). + * + * <p>Unlike {@link SavepointDataStreamScanProvider}, this reads state registered under an arbitrary + * namespace rather than {@code VoidNamespace}, so it drives {@link + * SavepointReader#readWindowKeyedState}. + */ +@SuppressWarnings("rawtypes") +public class WindowSavepointDataStreamScanProvider + extends AbstractMultiColumnScanProvider<WindowStateTableMapping> { + + public WindowSavepointDataStreamScanProvider( + @Nullable final String stateBackendType, + final String statePath, + final OperatorIdentifier operatorIdentifier, + final Supplier<WindowStateTableMapping> mappingSupplier, + RowType rowType, + @Nullable SavepointKeyFilter keyFilter) { + super(stateBackendType, statePath, operatorIdentifier, mappingSupplier, rowType, keyFilter); + } + + @Override + @SuppressWarnings("unchecked") + protected DataStream<RowData> readState( + SavepointReader savepointReader, WindowStateTableMapping mapping) throws Exception { + List<String> stateNames = + mapping.getAllValueColumns().stream() + .map(StateValueColumnConfiguration::getStateName) + .collect(Collectors.toList()); + + return savepointReader.readWindowKeyedState( + operatorIdentifier, + new WindowKeyedStateReader(rowType, mapping), + (TypeInformation) mapping.getKeyTypeInfo(), + (TypeSerializer) mapping.getNamespaceSerializer(), + stateNames, + InternalTypeInfo.of(rowType), + keyFilter); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowStateTableMapping.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowStateTableMapping.java new file mode 100644 index 00000000000..e48d03ae922 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/WindowStateTableMapping.java @@ -0,0 +1,235 @@ +/* + * 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.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.state.api.OperatorIdentifier; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +import static org.apache.flink.state.table.SavepointConnectorOptions.STATE_NAME; + +/** + * Maps the key column, the window (namespace) column and the state value columns to their positions + * in the output row, for the general namespaced (e.g. window-scoped) keyed state table. + * + * <p>The key and window columns are identified by name ({@link #KEY_COLUMN_NAME}, {@link + * #WINDOW_COLUMN_NAME}) rather than by primary key position: the table declares a composite primary + * key over both columns (see {@link org.apache.flink.state.api.StateTableUtils}), but {@link + * StateTableMapping#validateAndExtractKeyColumn} only supports a single-column primary key, so the + * window table resolves its columns by name instead of reusing that logic. + * + * <p>Projection push-down remaps column indices exactly as in {@link StateTableMapping}. + */ +@Internal +public class WindowStateTableMapping implements Serializable, MultiColumnStateMapping { + + private static final long serialVersionUID = 1L; + + public static final String KEY_COLUMN_NAME = "state_key"; + public static final String WINDOW_COLUMN_NAME = "state_window"; + + private final int keyColumnIndex; + private final int windowColumnIndex; + private final List<StateValueColumnConfiguration> valueColumns; + private final List<StateValueColumnConfiguration> allValueColumns; + + /** Resolved key type info; {@code null} when the mapping was not created via {@link #from}. */ + @Nullable private final TypeInformation<?> keyTypeInfo; + + /** + * Serializer of the namespace (e.g. window) under which the value states are registered; {@code + * null} when the mapping was not created via {@link #from}. + */ + @Nullable private final TypeSerializer<Object> namespaceSerializer; + + public WindowStateTableMapping( + int keyColumnIndex, + int windowColumnIndex, + List<StateValueColumnConfiguration> valueColumns) { + this(keyColumnIndex, windowColumnIndex, null, valueColumns, valueColumns, null); + } + + private WindowStateTableMapping( + int keyColumnIndex, + int windowColumnIndex, + @Nullable TypeInformation<?> keyTypeInfo, + List<StateValueColumnConfiguration> valueColumns, + List<StateValueColumnConfiguration> allValueColumns, + @Nullable TypeSerializer<Object> namespaceSerializer) { + this.keyColumnIndex = keyColumnIndex; + this.windowColumnIndex = windowColumnIndex; + this.keyTypeInfo = keyTypeInfo; + this.valueColumns = valueColumns; + this.allValueColumns = allValueColumns; + this.namespaceSerializer = namespaceSerializer; + } + + public int getKeyColumnIndex() { + return keyColumnIndex; + } + + public int getWindowColumnIndex() { + return windowColumnIndex; + } + + /** Columns to write to the output row (may be a projected subset). */ + public List<StateValueColumnConfiguration> getValueColumns() { + return valueColumns; + } + + @Override + public List<StateValueColumnConfiguration> getAllValueColumns() { + return allValueColumns; + } + + @Override + @Nullable + public TypeInformation<?> getKeyTypeInfo() { + return keyTypeInfo; + } + + @Nullable + public TypeSerializer<Object> getNamespaceSerializer() { + return namespaceSerializer; + } + + /** + * Creates a new {@link WindowStateTableMapping} with column indices remapped to the projected + * output. Only flat projections ({@code projectedFields[i].length == 1}) are supported. + */ + @Override + public WindowStateTableMapping project(int[][] projectedFields) { + return new WindowStateTableMapping( + TableMappingSupport.remapColumnIndex(projectedFields, this.keyColumnIndex), + TableMappingSupport.remapColumnIndex(projectedFields, this.windowColumnIndex), + keyTypeInfo, + TableMappingSupport.remapValueColumns(projectedFields, this.valueColumns), + allValueColumns, + namespaceSerializer); + } + + // ------------------------------------------------------------------------- + // Factory + // ------------------------------------------------------------------------- + + /** + * Validates the table schema, registers all per-field connector {@link ConfigOption}s into + * {@code optionalOptions} for option validation, and returns {@code {keyColumnIndex, + * windowColumnIndex}} in the physical row data type. + * + * <p>Unlike {@link StateTableMapping}, the key and window columns are identified by name + * ({@link #KEY_COLUMN_NAME}, {@link #WINDOW_COLUMN_NAME}) rather than by primary key position: + * the window table's primary key spans both columns, and {@link + * StateTableMapping#validateAndExtractKeyColumn}'s single-primary-key-column assumption doesn't + * apply here. + * + * <p>This is a purely structural operation: it analyses the table schema but performs no class + * loading or type resolution. Call this eagerly so that option validation passes, then wrap + * {@link #from} in a {@code Supplier} to defer class loading to scan time. + */ + public static int[] validateAndExtractKeyAndWindowColumns( + ResolvedCatalogTable catalogTable, Set<ConfigOption<?>> optionalOptions) { + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + int keyIdx = TableMappingSupport.columnIndex(physicalDataType, KEY_COLUMN_NAME); + int windowIdx = TableMappingSupport.columnIndex(physicalDataType, WINDOW_COLUMN_NAME); + if (keyIdx < 0 || windowIdx < 0) { + throw new ValidationException( + "Namespaced (window) keyed state tables must have a '" + + (keyIdx < 0 ? KEY_COLUMN_NAME : WINDOW_COLUMN_NAME) + + "' column."); + } + + RowType rowType = (RowType) physicalDataType.getLogicalType(); + for (int colIdx : + TableMappingSupport.valueColumnIndices(physicalDataType, keyIdx, windowIdx)) { + RowType.RowField valueRowField = rowType.getFields().get(colIdx); + optionalOptions.add( + TableMappingSupport.fieldOption(valueRowField.getName(), STATE_NAME) + .stringType() + .noDefaultValue()); + } + + return new int[] {keyIdx, windowIdx}; + } + + /** + * Builds a complete {@link WindowStateTableMapping} from a {@link ResolvedCatalogTable}, + * loading operator state metadata from the savepoint and resolving serializers, key type and + * namespace serializer from it. + * + * <p>Assumes {@link #validateAndExtractKeyAndWindowColumns} 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. + */ + public static WindowStateTableMapping from( + ResolvedCatalogTable catalogTable, + Configuration options, + String statePath, + OperatorIdentifier operatorIdentifier, + SerializerConfig serializerConfig) { + + SavepointTypeInfoResolver typeResolver = + TableMappingSupport.createTypeResolver( + statePath, operatorIdentifier, serializerConfig); + + DataType physicalDataType = catalogTable.getResolvedSchema().toPhysicalRowDataType(); + RowType rowType = (RowType) physicalDataType.getLogicalType(); + int keyIdx = TableMappingSupport.columnIndex(physicalDataType, KEY_COLUMN_NAME); + int windowIdx = TableMappingSupport.columnIndex(physicalDataType, WINDOW_COLUMN_NAME); + + TypeInformation<?> keyTypeInfo = + typeResolver.resolveKeyType(rowType.getFields().get(keyIdx)); + + List<StateValueColumnConfiguration> valueColumns = new ArrayList<>(); + for (int colIdx : + TableMappingSupport.valueColumnIndices(physicalDataType, keyIdx, windowIdx)) { + valueColumns.add( + TableMappingSupport.createValueColumnConfig( + colIdx, rowType, options, typeResolver)); + } + if (valueColumns.isEmpty()) { + throw new ValidationException( + "Namespaced (window) keyed state tables must have at least one state column."); + } + + @SuppressWarnings("unchecked") + TypeSerializer<Object> namespaceSerializer = + (TypeSerializer<Object>) + typeResolver.resolveNamespaceSerializer(valueColumns.get(0).getStateName()); + + return new WindowStateTableMapping( + keyIdx, windowIdx, keyTypeInfo, valueColumns, valueColumns, namespaceSerializer); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunctionTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunctionTest.java new file mode 100644 index 00000000000..61d6daa86a7 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/functions/WindowKeyedStateReaderFunctionTest.java @@ -0,0 +1,45 @@ +/* + * 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.functions; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.Arrays; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Guards against {@link WindowKeyedStateReaderFunction.Context} silently growing a timer accessor + * (unlike {@link KeyedStateReaderFunction.Context}, it exposes none by design: the operator has + * never supported timers). + */ +public class WindowKeyedStateReaderFunctionTest { + + @Test + public void testContextExposesOnlyGetState() { + Method[] methods = WindowKeyedStateReaderFunction.Context.class.getMethods(); + assertEquals( + 1, + methods.length, + "Unexpected method set on WindowKeyedStateReaderFunction.Context: " + + Arrays.toString(methods)); + assertEquals("getState", methods[0].getName()); + } +}
