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 5e16408ec3480df6b43f9b5773028c98b0b7dfae Author: Gyula Fora <[email protected]> AuthorDate: Sun Jul 19 15:09:53 2026 +0200 [FLINK-40177][state-processor-api] State table utils and type inference for keyed state Introduces the core utilities used to build keyed state catalog tables: StateSchemaExtractor reads the raw keyed-state serialization header to determine registered states without loading user POJO classes, SerializerSnapshotToLogicalTypeConverter maps TypeSerializerSnapshot trees to Flink SQL LogicalType, and StateTableUtils ties these together to classify an operator keyed states and build the corresponding CatalogTable, including best-effort state backend detection. Windowed, flattened, and non-keyed state support are intentionally left out and will be added in later commits. --- .../apache/flink/state/api/StateTableUtils.java | 501 +++++++++++++++++++++ .../flink/state/api/schema/AvroStateUtils.java | 121 +++++ .../state/api/schema/KeyedStateSchemaInfo.java | 90 ++++ .../SerializerSnapshotToLogicalTypeConverter.java | 230 ++++++++++ .../state/api/schema/StateSchemaExtractor.java | 141 ++++++ .../flink/state/api/schema/StateSchemaInfo.java | 82 ++++ .../EmbeddedRocksDBKeyedStateReadingITCase.java | 31 ++ .../state/api/HashMapKeyedStateReadingITCase.java | 31 ++ .../flink/state/api/KeyedStateReadingITCase.java | 111 +++++ .../flink/state/api/StateTableUtilsTest.java | 129 ++++++ ...rializerSnapshotToLogicalTypeConverterTest.java | 425 +++++++++++++++++ .../state/api/schema/StateSchemaExtractorTest.java | 204 +++++++++ 12 files changed, 2096 insertions(+) diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java new file mode 100644 index 00000000000..7b681c88a8a --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java @@ -0,0 +1,501 @@ +/* + * 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; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle; +import org.apache.flink.runtime.state.KeyGroupsSavepointStateHandle; +import org.apache.flink.runtime.state.KeyGroupsStateHandle; +import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.StateBackendLoader; +import org.apache.flink.runtime.state.VoidNamespaceSerializer; +import org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle; +import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter; +import org.apache.flink.state.api.schema.StateSchemaExtractor; +import org.apache.flink.state.api.schema.StateSchemaInfo; +import org.apache.flink.state.table.SavepointConnectorOptions; +import org.apache.flink.streaming.api.operators.InternalTimeServiceManagerImpl; +import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.VarBinaryType; +import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * High-level utility for inspecting and reading keyed state from a checkpoint / savepoint without + * requiring user POJO classes on the classpath. + */ +@Internal +public final class StateTableUtils { + + private static final Logger LOG = LoggerFactory.getLogger(StateTableUtils.class); + + private StateTableUtils() {} + + /** + * Returns the {@link OperatorIdentifier}s of all operators present in the given checkpoint + * metadata that have at least one non-internal keyed state. + * + * @param metadata the checkpoint metadata to inspect + * @return list of operator identifiers; never null, may be empty + */ + public static List<OperatorIdentifier> getOperatorIdentifiers(CheckpointMetadata metadata) { + return metadata.getOperatorStates().stream() + .filter(StateTableUtils::hasNonInternalKeyedState) + .map( + op -> + op.getOperatorUid() + .map(OperatorIdentifier::forUid) + .orElseGet( + () -> + OperatorIdentifier.forUidHash( + op.getOperatorID().toHexString()))) + .collect(Collectors.toList()); + } + + private static boolean hasNonInternalKeyedState(OperatorState op) { + try { + List<StateSchemaInfo> schemas = StateSchemaExtractor.extractSchema(op); + ClassifiedStates classified = classifyStates(op.getOperatorID().toHexString(), schemas); + return !classified.voidNamespaceStates.isEmpty() + || !classified.windowNamespaceStates.isEmpty(); + } catch (Exception e) { + LOG.error( + "Could not extract state schema for operator '{}': {}. Excluding from catalog.", + op.getOperatorID(), + e.getMessage()); + return false; + } + } + + /** + * Returns the names of all keyed states registered by the given operator. + * + * @param metadata the checkpoint metadata to inspect + * @param operatorId identifies the operator + * @param classLoader the class loader used when reading serializer snapshots + * @return list of state names; never null, may be empty + * @throws IOException if the state header cannot be read + */ + public static List<String> getKeyedStates( + CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { + OperatorState opState = findOperatorState(metadata, operatorId); + List<StateSchemaInfo> schemaInfos = StateSchemaExtractor.extractSchema(opState); + ClassifiedStates classified = classifyStates(operatorId.toString(), schemaInfos); + return classified.voidNamespaceStates.stream() + .map(info -> info.stateName) + .collect(Collectors.toList()); + } + + /** + * Returns the {@link KeyedStateSchemaInfo} for the plain per-key (void-namespace) states of the + * given operator — the ones exposed by the {@code _keyed}/{@code _keyed_flat} tables. + * + * <p>Schema extraction is lenient: POJO field names and types are derived from the serializer + * snapshot and do not require the user POJO class to be on the classpath. + * + * @param metadata the checkpoint metadata to inspect + * @param operatorId identifies the operator + * @return schema information covering the key type and all registered state entries + * @throws IOException if the state header cannot be read + */ + public static KeyedStateSchemaInfo getKeyedStateSchema( + CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException { + OperatorState opState = findOperatorState(metadata, operatorId); + List<StateSchemaInfo> schemas = StateSchemaExtractor.extractSchema(opState); + ClassifiedStates classified = classifyStates(operatorId.toString(), schemas); + return buildKeyedStateSchemaInfo(schemas, classified.voidNamespaceStates, null); + } + + private static KeyedStateSchemaInfo buildKeyedStateSchemaInfo( + List<StateSchemaInfo> allSchemas, + List<StateSchemaInfo> statesToInclude, + @Nullable LogicalType windowLogicalType) { + LogicalType keyType = + allSchemas.isEmpty() + ? new VarBinaryType(true, VarBinaryType.MAX_LENGTH) + : SerializerSnapshotToLogicalTypeConverter.convert( + allSchemas.get(0).keySnapshot); + + LinkedHashMap<String, KeyedStateSchemaInfo.StateEntryInfo> stateSchemas = + new LinkedHashMap<>(); + for (StateSchemaInfo info : statesToInclude) { + SavepointConnectorOptions.StateType stateType; + if (info.stateKind == StateDescriptor.Type.LIST) { + stateType = SavepointConnectorOptions.StateType.LIST; + } else if (info.stateKind == StateDescriptor.Type.MAP) { + stateType = SavepointConnectorOptions.StateType.MAP; + } else { + stateType = SavepointConnectorOptions.StateType.VALUE; + } + + try { + LogicalType logicalType = + SerializerSnapshotToLogicalTypeConverter.convert(info.valueSnapshot); + stateSchemas.put( + info.stateName, + new KeyedStateSchemaInfo.StateEntryInfo( + stateType, logicalType, windowLogicalType)); + } catch (Exception e) { + logSchemaExtractionFailure("", info.stateName, info.valueSnapshot, e); + } + } + + return new KeyedStateSchemaInfo(keyType, stateSchemas); + } + + /** + * Logs that a single state's schema could not be extracted and will therefore be excluded from + * the table schema, shared by {@link #buildKeyedStateSchemaInfo}. + * + * @param label a prefix inserted before "state" in the log message (e.g. {@code "non-keyed "} + * or {@code ""}), distinguishing which caller excluded the state + */ + private static void logSchemaExtractionFailure( + String label, + String stateName, + @Nullable TypeSerializerSnapshot<?> valueSnapshot, + Exception e) { + LOG.error( + "Cannot extract schema for {}state '{}' (serializer type: {}): {}. " + + "This state will be excluded from the table schema. " + + "Use explicit connector options to include it.", + label, + stateName, + valueSnapshot == null ? "null" : valueSnapshot.getClass().getSimpleName(), + e.getMessage()); + } + + /** + * Builds a {@link CatalogTable} representing all keyed states of an operator. + * + * <p>The resulting table has one column named {@code "state_key"} for the key and one column + * per keyed state. The connector options are pre-populated so the table can be registered + * directly in a {@link org.apache.flink.table.catalog.CatalogManager}. + * + * <p>When the state backend that produced the operator's keyed state can be unambiguously + * determined from the checkpoint metadata, {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} + * is pre-populated as well, so callers don't need to specify it themselves. + * + * @param metadata the checkpoint metadata the operator belongs to + * @param schemaInfo the schema information returned by {@link #getKeyedStateSchema} + * @param statePath the path to the savepoint / checkpoint + * @param operatorIdentifier identifies the operator whose state to read + * @return a {@link CatalogTable} ready for registration + */ + public static CatalogTable getStateCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String statePath, + OperatorIdentifier operatorIdentifier) { + return buildKeyedCatalogTable(metadata, schemaInfo, statePath, operatorIdentifier, null); + } + + /** + * Builds a {@link CatalogTable} representing all keyed states of an operator, or, when {@code + * windowType} is non-null, all namespaced (e.g. window-scoped) states of an operator. + */ + private static CatalogTable buildKeyedCatalogTable( + CheckpointMetadata metadata, + KeyedStateSchemaInfo schemaInfo, + String statePath, + OperatorIdentifier operatorIdentifier, + @Nullable LogicalType windowType) { + + Schema.Builder schemaBuilder = Schema.newBuilder(); + schemaBuilder.column( + "state_key", LogicalTypeDataTypeConverter.toDataType(schemaInfo.keyType).notNull()); + if (windowType != null) { + schemaBuilder.column( + "state_window", LogicalTypeDataTypeConverter.toDataType(windowType).notNull()); + } + + for (Map.Entry<String, KeyedStateSchemaInfo.StateEntryInfo> entry : + schemaInfo.stateSchemas.entrySet()) { + schemaBuilder.column(entry.getKey(), stateValueColumnDataType(entry.getValue())); + } + if (windowType == null) { + schemaBuilder.primaryKeyNamed("PK_state_key", "state_key"); + } + Schema schema = schemaBuilder.build(); + + Map<String, String> options = buildBaseConnectorOptions(statePath, operatorIdentifier); + withStateBackendType(options, metadata, operatorIdentifier); + + return CatalogTable.newBuilder().schema(schema).options(options).build(); + } + + // ------------------------------------------------------------------------- + // Private helpers + // ------------------------------------------------------------------------- + + /** + * Resolves the SQL column {@link org.apache.flink.table.types.DataType} for a single state's + * value column, forcing it nullable for VALUE-shaped state: unlike LIST/MAP (which always have + * a value, possibly empty), a {@code ValueState}/{@code ReducingState}/{@code AggregatingState} + * can legitimately hold no value (e.g. never written, or cleared by a trigger such as {@code + * CountTrigger}), in which case a read returns {@code null}. + */ + private static DataType stateValueColumnDataType( + KeyedStateSchemaInfo.StateEntryInfo entryInfo) { + DataType dataType = LogicalTypeDataTypeConverter.toDataType(entryInfo.logicalType); + return entryInfo.stateType == SavepointConnectorOptions.StateType.VALUE + ? dataType.nullable() + : dataType; + } + + private static OperatorState findOperatorState( + CheckpointMetadata metadata, OperatorIdentifier operatorId) { + for (OperatorState op : metadata.getOperatorStates()) { + if (op.getOperatorID().equals(operatorId.getOperatorId())) { + return op; + } + } + throw new IllegalArgumentException( + "Operator '" + operatorId + "' not found in checkpoint metadata."); + } + + /** + * Returns the base connector options ({@link FactoryUtil#CONNECTOR}, {@link + * SavepointConnectorOptions#STATE_PATH}, and the operator identifier option) shared by every + * savepoint-backed {@link CatalogTable}. + */ + private static Map<String, String> buildBaseConnectorOptions( + String statePath, OperatorIdentifier operatorIdentifier) { + Map<String, String> options = new HashMap<>(); + options.put(FactoryUtil.CONNECTOR.key(), "savepoint"); + options.put(SavepointConnectorOptions.STATE_PATH.key(), statePath); + operatorIdentifier + .getUid() + .ifPresentOrElse( + uid -> options.put(SavepointConnectorOptions.OPERATOR_UID.key(), uid), + () -> + options.put( + SavepointConnectorOptions.OPERATOR_UID_HASH.key(), + operatorIdentifier.getOperatorId().toHexString())); + return options; + } + + /** + * Adds {@link SavepointConnectorOptions#STATE_BACKEND_TYPE} to {@code options} when it can be + * unambiguously determined from the checkpoint metadata. Only meaningful for keyed state + * tables: non-keyed (list/union/broadcast) state isn't stored in a state backend, so callers + * for those table kinds must not call this. + */ + private static void withStateBackendType( + Map<String, String> options, + CheckpointMetadata metadata, + OperatorIdentifier operatorIdentifier) { + OperatorState opState = findOperatorState(metadata, operatorIdentifier); + detectStateBackendType(opState) + .ifPresent( + type -> + options.put( + SavepointConnectorOptions.STATE_BACKEND_TYPE.key(), type)); + } + + /** + * Attempts to determine the state backend (shortcut name, see {@link + * StateBackendLoader#HASHMAP_STATE_BACKEND_NAME} / {@link + * StateBackendLoader#ROCKSDB_STATE_BACKEND_NAME}) that produced the operator's keyed state, by + * inspecting the concrete {@link KeyedStateHandle} subtype found in the checkpoint metadata: + * heap/HashMap backends produce {@link KeyGroupsStateHandle}, RocksDB/ForSt backends produce + * {@link IncrementalKeyedStateHandle}. + * + * <p>Canonical-format savepoints rewrite keyed state into the backend-agnostic {@link + * KeyGroupsSavepointStateHandle}, in which case the originating backend can no longer be + * determined from the handle alone; an empty result is returned rather than guessing. + */ + static Optional<String> detectStateBackendType(OperatorState opState) { + Set<String> detectedTypes = new HashSet<>(); + for (OperatorSubtaskState subtaskState : opState.getStates()) { + collectStateBackendTypes(subtaskState.getManagedKeyedState(), detectedTypes); + collectStateBackendTypes(subtaskState.getRawKeyedState(), detectedTypes); + } + if (detectedTypes.size() != 1) { + if (detectedTypes.size() > 1) { + LOG.warn( + "Operator '{}' has keyed state handles from multiple state backends {}; " + + "not setting '{}'.", + opState.getOperatorID(), + detectedTypes, + SavepointConnectorOptions.STATE_BACKEND_TYPE.key()); + } + return Optional.empty(); + } + return Optional.of(detectedTypes.iterator().next()); + } + + private static void collectStateBackendTypes( + Iterable<KeyedStateHandle> handles, Set<String> detectedTypes) { + for (KeyedStateHandle handle : handles) { + if (handle instanceof ChangelogStateBackendHandle) { + collectStateBackendTypes( + ((ChangelogStateBackendHandle) handle).getMaterializedStateHandles(), + detectedTypes); + } else if (handle instanceof IncrementalKeyedStateHandle) { + detectedTypes.add(StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME); + } else if (handle instanceof KeyGroupsSavepointStateHandle) { + // Canonical-format savepoints rewrite keyed state into a backend-agnostic + // format; the originating backend can no longer be told apart from the handle. + } else if (handle instanceof KeyGroupsStateHandle) { + detectedTypes.add(StateBackendLoader.HASHMAP_STATE_BACKEND_NAME); + } else { + LOG.warn("Unknown handle type '{}'.", handle.getClass().getSimpleName()); + } + } + } + + /** Returns {@code true} for Flink-internal states that are not user-registered states. */ + private static boolean isInternalState(String stateName) { + return stateName.startsWith(InternalTimeServiceManagerImpl.TIMER_STATE_PREFIX + "/") + || stateName.equals(WindowOperator.MERGING_WINDOW_SET_STATE_NAME); + } + + /** + * Returns {@code true} if a state is plain per-key state (registered with {@code + * VoidNamespace}) and {@code false} if it is scoped by some other namespace (e.g. a window). + * + * <p>A missing namespace snapshot (e.g. from an older savepoint format) is treated as void, + * matching pre-existing behavior. + */ + private static boolean isVoidNamespace(TypeSerializerSnapshot<?> namespaceSnapshot) { + return namespaceSnapshot == null + || namespaceSnapshot + instanceof VoidNamespaceSerializer.VoidNamespaceSerializerSnapshot; + } + + /** + * The result of {@link #classifyStates}: user-registered states of an operator, partitioned + * into plain per-key (void-namespace) states and namespaced (e.g. window-scoped) states. + */ + private static final class ClassifiedStates { + final List<StateSchemaInfo> voidNamespaceStates; + final List<StateSchemaInfo> windowNamespaceStates; + @Nullable final LogicalType windowLogicalType; + + ClassifiedStates( + List<StateSchemaInfo> voidNamespaceStates, + List<StateSchemaInfo> windowNamespaceStates, + @Nullable LogicalType windowLogicalType) { + this.voidNamespaceStates = voidNamespaceStates; + this.windowNamespaceStates = windowNamespaceStates; + this.windowLogicalType = windowLogicalType; + } + } + + /** + * Partitions the user-registered states of a single operator into plain per-key + * (void-namespace) states and namespaced states, resolving the namespaced states' shared {@link + * LogicalType} along the way. + * + * <p>An operator may register states under more than one distinct namespace <em>type</em> only + * via hand-rolled state access (no built-in windowing API does this); when that happens, the + * first namespace type whose schema can be determined is kept, and every other group is + * excluded with a logged warning. + * + * @param operatorLabel a human-readable operator identifier, used only for log messages + */ + private static ClassifiedStates classifyStates( + String operatorLabel, List<StateSchemaInfo> schemas) { + List<StateSchemaInfo> voidStates = new ArrayList<>(); + Map<String, List<StateSchemaInfo>> namespacedGroups = new LinkedHashMap<>(); + for (StateSchemaInfo info : schemas) { + if (isInternalState(info.stateName)) { + continue; + } + if (isVoidNamespace(info.namespaceSnapshot)) { + voidStates.add(info); + } else { + namespacedGroups + .computeIfAbsent( + info.namespaceSnapshot.getClass().getName(), k -> new ArrayList<>()) + .add(info); + } + } + + List<StateSchemaInfo> chosenGroup = Collections.emptyList(); + LogicalType chosenNamespaceType = null; + for (Map.Entry<String, List<StateSchemaInfo>> entry : namespacedGroups.entrySet()) { + if (chosenNamespaceType != null) { + logExcludedNamespaceGroup( + operatorLabel, + entry.getKey(), + entry.getValue(), + "an operator has states registered under more than one namespace type"); + continue; + } + + TypeSerializerSnapshot<?> representative = entry.getValue().get(0).namespaceSnapshot; + try { + chosenNamespaceType = + SerializerSnapshotToLogicalTypeConverter.convert(representative); + } catch (UnsupportedOperationException e) { + logExcludedNamespaceGroup( + operatorLabel, + entry.getKey(), + entry.getValue(), + "cannot extract schema for this namespace type: " + e.getMessage()); + continue; + } + chosenGroup = entry.getValue(); + } + + return new ClassifiedStates(voidStates, chosenGroup, chosenNamespaceType); + } + + private static void logExcludedNamespaceGroup( + String operatorLabel, + String namespaceClassName, + List<StateSchemaInfo> excluded, + String reason) { + LOG.warn( + "Excluding namespace type '{}' on operator '{}' from the catalog: {}. States: {}.", + namespaceClassName, + operatorLabel, + reason, + excluded.stream().map(i -> i.stateName).collect(Collectors.toList())); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/AvroStateUtils.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/AvroStateUtils.java new file mode 100644 index 00000000000..df5de4c2fc9 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/AvroStateUtils.java @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter; +import org.apache.flink.formats.avro.typeutils.AvroSerializer; +import org.apache.flink.formats.avro.typeutils.AvroSerializerSnapshot; +import org.apache.flink.table.types.logical.LogicalType; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * All Avro-specific logic used by the State Processing API's schema-based table access, gathered in + * one place so that the classes which dispatch to it never need to mention an Avro type themselves. + * + * <p>flink-avro is an optional dependency of this module (see the module {@code pom.xml}). A class + * that mentions an Avro type directly - even in an {@code instanceof} branch that is never taken - + * makes the JVM try to resolve that type the moment the check is reached for *any* value, throwing + * {@code NoClassDefFoundError} for callers who never use Avro at all. Callers must therefore select + * the Avro-specific branch by class/interface name (see {@link + * #AVRO_SERIALIZER_SNAPSHOT_CLASS_NAME} and {@link #isGenericRecord}) and only then delegate here; + * this class is consequently only ever loaded once that name comparison has already confirmed Avro + * is genuinely on the classpath. + */ +@Internal +public final class AvroStateUtils { + + /** Fully-qualified name of {@code AvroSerializerSnapshot}, for dispatch by class name. */ + public static final String AVRO_SERIALIZER_SNAPSHOT_CLASS_NAME = + "org.apache.flink.formats.avro.typeutils.AvroSerializerSnapshot"; + + private static final String GENERIC_RECORD_CLASS_NAME = "org.apache.avro.generic.GenericRecord"; + + /** + * Memoizes {@link #isGenericRecord}: it walks a class's full interface hierarchy, so callers + * that check the same class repeatedly (e.g. once per field of every row of the same type) + * would otherwise repeat that walk every time. A plain static map is safe here since the answer + * is a pure function of the {@code Class} object - it never changes for a given class, and is + * shared happily across every {@link AvroStateUtils} caller and instance. + */ + private static final Map<Class<?>, Boolean> GENERIC_RECORD_CACHE = new ConcurrentHashMap<>(); + + private AvroStateUtils() {} + + /** + * Builds the fallback serializer for an {@code AvroSerializerSnapshot} with a missing class. + */ + public static TypeSerializer<?> createFallbackSerializer(TypeSerializerSnapshot<?> snapshot) { + Schema schema = ((AvroSerializerSnapshot<?>) snapshot).getSchema(); + return new AvroSerializer<>(GenericRecord.class, schema); + } + + /** + * Converts an {@code AvroSerializerSnapshot}'s embedded writer schema into a {@link + * LogicalType}. + */ + public static LogicalType convertToLogicalType(TypeSerializerSnapshot<?> snapshot) { + // getSchema() returns the writer schema embedded in the snapshot — always present + // regardless of whether the specific record class is on the classpath. + AvroSerializerSnapshot<?> avroSnapshot = (AvroSerializerSnapshot<?>) snapshot; + return AvroSchemaConverter.convertToDataType(avroSnapshot.getSchema().toString()) + .getLogicalType(); + } + + /** + * Returns {@code true} if {@code clazz}, or any class/interface in its hierarchy, is named + * {@code org.apache.avro.generic.GenericRecord}. Safe to call even when {@code GenericRecord} + * itself is not on the classpath: {@code clazz} could only have been loaded and instantiated if + * all interfaces it declares were already resolved, so walking {@link Class#getInterfaces()} + * never triggers a fresh classload of {@code GenericRecord}. + */ + public static boolean isGenericRecord(Class<?> clazz) { + return GENERIC_RECORD_CACHE.computeIfAbsent(clazz, AvroStateUtils::computeIsGenericRecord); + } + + /** + * Does the actual interface-hierarchy walk for {@link #isGenericRecord}. Recurses into itself + * rather than back into {@link #isGenericRecord}: {@code ConcurrentHashMap.computeIfAbsent} + * forbids its mapping function from calling back into the same map - even for a different key - + * and will throw {@code IllegalStateException("Recursive update")} if it does. + */ + private static boolean computeIsGenericRecord(Class<?> clazz) { + for (Class<?> current = clazz; current != null; current = current.getSuperclass()) { + for (Class<?> iface : current.getInterfaces()) { + if (iface.getName().equals(GENERIC_RECORD_CLASS_NAME) + || computeIsGenericRecord(iface)) { + return true; + } + } + } + return false; + } + + /** Reads a field from an Avro {@code GenericRecord}. */ + public static Object getGenericRecordField(Object record, String fieldName) { + return ((GenericRecord) record).get(fieldName); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/KeyedStateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/KeyedStateSchemaInfo.java new file mode 100644 index 00000000000..014a89dc018 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/KeyedStateSchemaInfo.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.state.table.SavepointConnectorOptions; +import org.apache.flink.table.types.logical.LogicalType; + +import javax.annotation.Nullable; + +import java.util.LinkedHashMap; + +/** + * Schema information for all keyed states of a single operator, extracted from a savepoint without + * requiring user POJO classes on the classpath. + */ +@Internal +public final class KeyedStateSchemaInfo { + + /** The logical type of the keyed state backend key (e.g., BigIntType for Long keys). */ + public final LogicalType keyType; + + /** + * Ordered map of registered state names to their entry information. Ordered by the registration + * order found in the savepoint. + */ + public final LinkedHashMap<String, StateEntryInfo> stateSchemas; + + public KeyedStateSchemaInfo( + LogicalType keyType, LinkedHashMap<String, StateEntryInfo> stateSchemas) { + this.keyType = keyType; + this.stateSchemas = stateSchemas; + } + + /** Schema information for one keyed state entry. */ + public static final class StateEntryInfo { + + /** VALUE, LIST, or MAP. */ + public final SavepointConnectorOptions.StateType stateType; + + /** + * The SQL column logical type. + * + * <ul> + * <li>VALUE<Long>: BigIntType + * <li>VALUE<POJO>: RowType (field names + types from the serializer snapshot) + * <li>LIST<Long>: ArrayType(BigIntType) + * <li>MAP<Long,Long>: MapType(BigIntType, BigIntType) + * </ul> + */ + public final LogicalType logicalType; + + /** + * The resolved {@link LogicalType} of the state's namespace, or {@code null} if the state + * is plain per-key state (registered under {@code VoidNamespace}). Non-null means the state + * is scoped by some other namespace (e.g. a window), resolved generically by {@link + * SerializerSnapshotToLogicalTypeConverter} rather than as a fixed {@code + * TimeWindow}-shaped type. + * + * <p>Named "window" rather than "namespace" because this is the user-facing, + * post-conversion form; see {@link StateSchemaInfo} for the raw/resolved naming convention. + */ + @Nullable public final LogicalType windowLogicalType; + + public StateEntryInfo( + SavepointConnectorOptions.StateType stateType, + LogicalType logicalType, + @Nullable LogicalType windowLogicalType) { + this.stateType = stateType; + this.logicalType = logicalType; + this.windowLogicalType = windowLogicalType; + } + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverter.java new file mode 100644 index 00000000000..b689cbe9656 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverter.java @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.ByteSerializer; +import org.apache.flink.api.common.typeutils.base.CharSerializer; +import org.apache.flink.api.common.typeutils.base.DoubleSerializer; +import org.apache.flink.api.common.typeutils.base.EnumSerializer; +import org.apache.flink.api.common.typeutils.base.FloatSerializer; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.LongSerializer; +import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.ShortSerializer; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.java.typeutils.runtime.NullableSerializer.NullableSerializerSnapshot; +import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializerSnapshot; +import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; +import org.apache.flink.streaming.api.windowing.windows.TimeWindow; +import org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.BooleanType; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.DoubleType; +import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.SmallIntType; +import org.apache.flink.table.types.logical.TimestampType; +import org.apache.flink.table.types.logical.TinyIntType; +import org.apache.flink.table.types.logical.VarBinaryType; +import org.apache.flink.table.types.logical.VarCharType; + +import java.util.AbstractMap; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Converts a {@link TypeSerializerSnapshot} tree into a Flink SQL {@link LogicalType}. + * + * <p>This is a pure function — no I/O, no class loading beyond what the snapshot itself already + * contains. Field names are extracted from {@link PojoSerializerSnapshot} entries; all other + * composite types use positional names {@code f0, f1, …}. + * + * <p>flink-avro is an optional dependency of this module (see the module {@code pom.xml}), so this + * class must never mention an Avro type directly: doing so - even in an {@code instanceof} branch - + * would make the JVM try to resolve that type the moment this method is reached for *any* + * unrecognized snapshot (e.g. a plain {@code ListSerializerSnapshot}), throwing {@code + * NoClassDefFoundError} for callers who never use Avro at all. The Avro-specific branch is instead + * selected by class name and delegated to {@link AvroStateUtils}, which is only ever loaded once + * that name comparison has already confirmed Avro is genuinely on the classpath. + */ +@Internal +public final class SerializerSnapshotToLogicalTypeConverter { + + /** Type used for snapshots we cannot describe any further, e.g. a missing nested snapshot. */ + private static final LogicalType OPAQUE_TYPE = + new VarBinaryType(true, VarBinaryType.MAX_LENGTH); + + private static final TypeSerializerSnapshot<?>[] NO_NESTED_SNAPSHOTS = + new TypeSerializerSnapshot<?>[0]; + + /** + * Snapshot classes that always map to the same {@link LogicalType}. Matched by exact class + * because every one of them is final; {@link LogicalType} instances are immutable and therefore + * safe to share. + * + * <p>The window entries are not namespace-specific: they fire identically if a window type is + * ever used as an ordinary value rather than as a namespace. + */ + private static final Map<Class<?>, LogicalType> FIXED_TYPES = createFixedTypes(); + + private SerializerSnapshotToLogicalTypeConverter() {} + + private static Map<Class<?>, LogicalType> createFixedTypes() { + Map<Class<?>, LogicalType> types = new HashMap<>(); + types.put(IntSerializer.IntSerializerSnapshot.class, new IntType(false)); + types.put(LongSerializer.LongSerializerSnapshot.class, new BigIntType(false)); + types.put(FloatSerializer.FloatSerializerSnapshot.class, new FloatType(false)); + types.put(DoubleSerializer.DoubleSerializerSnapshot.class, new DoubleType(false)); + types.put(BooleanSerializer.BooleanSerializerSnapshot.class, new BooleanType(false)); + types.put(ByteSerializer.ByteSerializerSnapshot.class, new TinyIntType(false)); + types.put(ShortSerializer.ShortSerializerSnapshot.class, new SmallIntType(false)); + types.put(CharSerializer.CharSerializerSnapshot.class, new CharType(false, 1)); + types.put( + StringSerializer.StringSerializerSnapshot.class, + new VarCharType(true, VarCharType.MAX_LENGTH)); + types.put( + TimeWindow.Serializer.TimeWindowSerializerSnapshot.class, + new RowType( + false, + List.of( + new RowType.RowField("window_start", new TimestampType(false, 3)), + new RowType.RowField("window_end", new TimestampType(false, 3))))); + types.put( + GlobalWindow.Serializer.GlobalWindowSerializerSnapshot.class, + new RowType(false, List.of())); + return types; + } + + /** + * Converts the given snapshot to a {@link LogicalType}. + * + * @param snapshot the serializer snapshot to convert, may be null + * @return a {@link LogicalType} corresponding to the snapshot type + * @throws UnsupportedOperationException if the snapshot type is not supported for schema-based + * table access + */ + public static LogicalType convert(TypeSerializerSnapshot<?> snapshot) { + if (snapshot == null) { + return OPAQUE_TYPE; + } + + LogicalType fixedType = FIXED_TYPES.get(snapshot.getClass()); + if (fixedType != null) { + return fixedType; + } + + if (snapshot instanceof PojoSerializerSnapshot) { + return convertPojo((PojoSerializerSnapshot<?>) snapshot); + } + if (snapshot instanceof EnumSerializer.EnumSerializerSnapshot) { + return new VarCharType(true, VarCharType.MAX_LENGTH); + } + String avroSnapshotClassName = AvroStateUtils.AVRO_SERIALIZER_SNAPSHOT_CLASS_NAME; + if (avroSnapshotClassName.equals(snapshot.getClass().getName())) { + return AvroStateUtils.convertToLogicalType(snapshot); + } + if (snapshot instanceof ListSerializerSnapshot) { + return new ArrayType(true, convertNested(snapshot, 0)); + } + if (snapshot instanceof MapSerializerSnapshot) { + return new MapType(true, convertNested(snapshot, 0), convertNested(snapshot, 1)); + } + if (snapshot instanceof NullableSerializerSnapshot) { + return convertNested(snapshot, 0).copy(true); + } + if (snapshot instanceof TupleSerializerSnapshot) { + return convertTuple((TupleSerializerSnapshot<?>) snapshot); + } + if (snapshot instanceof RowDataSerializerSnapshot) { + return convertRowData((RowDataSerializerSnapshot) snapshot); + } + + throw new UnsupportedOperationException( + "Cannot extract schema for TypeSerializerSnapshot of type '" + + snapshot.getClass().getName() + + "'. This serializer type is not supported for schema-based table" + + " access."); + } + + private static LogicalType convertPojo(PojoSerializerSnapshot<?> snapshot) { + List<AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>> fieldEntries = + snapshot.getFieldSnapshotEntries(); + List<RowType.RowField> fields = new ArrayList<>(fieldEntries.size()); + for (AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>> entry : fieldEntries) { + fields.add(new RowType.RowField(entry.getKey(), convert(entry.getValue()))); + } + return new RowType(true, fields); + } + + private static LogicalType convertTuple(TupleSerializerSnapshot<?> snapshot) { + TypeSerializerSnapshot<?>[] nested = nestedSnapshots(snapshot); + List<RowType.RowField> fields = new ArrayList<>(nested.length); + for (int i = 0; i < nested.length; i++) { + fields.add(new RowType.RowField("f" + i, convert(nested[i]))); + } + return new RowType(true, fields); + } + + /** + * Converts a {@link RowDataSerializerSnapshot} from its {@link + * RowDataSerializerSnapshot#getTypes()} instead of walking its nested field snapshots: a {@link + * LogicalType} is already a complete, self-describing schema (including nested field names), + * unlike the POJO/Avro case where the snapshot tree is the only source of field names. + */ + private static LogicalType convertRowData(RowDataSerializerSnapshot snapshot) { + LogicalType[] types = snapshot.getTypes(); + String[] fieldNames = snapshot.getFieldNames(); + List<RowType.RowField> fields = new ArrayList<>(types.length); + for (int i = 0; i < types.length; i++) { + String fieldName = fieldNames != null ? fieldNames[i] : "f" + i; + fields.add(new RowType.RowField(fieldName, types[i])); + } + return new RowType(true, fields); + } + + private static LogicalType convertNested(TypeSerializerSnapshot<?> snapshot, int index) { + TypeSerializerSnapshot<?>[] nested = nestedSnapshots(snapshot); + return convert(index < nested.length ? nested[index] : null); + } + + private static TypeSerializerSnapshot<?>[] nestedSnapshots(TypeSerializerSnapshot<?> snapshot) { + if (snapshot instanceof CompositeTypeSerializerSnapshot) { + TypeSerializerSnapshot<?>[] nested = + ((CompositeTypeSerializerSnapshot<?, ?>) snapshot) + .getNestedSerializerSnapshots(); + if (nested != null) { + return nested; + } + } + return NO_NESTED_SNAPSHOTS; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java new file mode 100644 index 00000000000..c58c7bf7737 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaExtractor.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.state.IncrementalKeyedStateHandle; +import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; +import org.apache.flink.runtime.state.KeyedStateHandle; +import org.apache.flink.runtime.state.StreamStateHandle; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.CommonOptionsKeys; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.CommonSerializerKeys; +import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Utility for extracting {@link StateSchemaInfo} from a savepoint without instantiating the full + * state backend or requiring user POJO classes on the classpath. + * + * <p>It reads the {@link KeyedBackendSerializationProxy} header that every heap/RocksDB keyed state + * file starts with. + */ +@Internal +public final class StateSchemaExtractor { + + private StateSchemaExtractor() {} + + /** + * Reads state schema information from the first available keyed state handle in the given + * operator state. + * + * <p>Returns an empty list rather than throwing when no keyed state handle is found: an + * operator may register only non-keyed (list/union/broadcast) state, in which case it has no + * keyed state to describe. + * + * <p>The metadata header lives in different places depending on the state backend: heap ({@code + * HashMapStateBackend}) savepoints hand back a {@code KeyGroupsStateHandle}, which is itself a + * {@link StreamStateHandle} starting with the header; RocksDB (incremental or full native) + * snapshots hand back an {@link IncrementalKeyedStateHandle}, whose own data stream starts with + * the SST payload instead, so the header must be read from {@link + * IncrementalKeyedStateHandle#getMetaDataStateHandle()}. + * + * @param operatorState the operator state from a loaded savepoint / checkpoint metadata + * @return list of schema info, one entry per registered state; never null, may be empty + * @throws IOException if the state header cannot be read + */ + public static List<StateSchemaInfo> extractSchema(OperatorState operatorState) + throws IOException { + + for (OperatorSubtaskState subtask : operatorState.getSubtaskStates().values()) { + for (KeyedStateHandle handle : subtask.getManagedKeyedState()) { + StreamStateHandle metadataHandle = null; + if (handle instanceof IncrementalKeyedStateHandle) { + metadataHandle = + ((IncrementalKeyedStateHandle) handle).getMetaDataStateHandle(); + } else if (handle instanceof StreamStateHandle) { + metadataHandle = (StreamStateHandle) handle; + } + if (metadataHandle != null) { + try (java.io.InputStream stream = metadataHandle.openInputStream()) { + return extractSchema(new DataInputViewStreamWrapper(stream)); + } + } + } + } + return Collections.emptyList(); + } + + /** + * Package-private overload that accepts a {@link DataInputView} directly. Allows unit tests to + * inject pre-built byte arrays without a real filesystem. + */ + static List<StateSchemaInfo> extractSchema(DataInputView in) throws IOException { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + KeyedBackendSerializationProxy<?> proxy = new KeyedBackendSerializationProxy<>(classLoader); + CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create); + proxy.read(in); + + TypeSerializerSnapshot<?> keySnapshot = proxy.getKeySerializerSnapshot(); + List<StateSchemaInfo> result = new ArrayList<>(); + + for (StateMetaInfoSnapshot meta : proxy.getStateMetaInfoSnapshots()) { + String kindStr = meta.getOption(CommonOptionsKeys.KEYED_STATE_TYPE); + StateDescriptor.Type stateKind; + try { + stateKind = StateDescriptor.Type.valueOf(kindStr); + } catch (IllegalArgumentException | NullPointerException e) { + stateKind = StateDescriptor.Type.UNKNOWN; + } + + TypeSerializerSnapshot<?> valueSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER); + TypeSerializerSnapshot<?> mapKeySnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.USER_KEY_SERIALIZER); + TypeSerializerSnapshot<?> namespaceSnapshot = + meta.getTypeSerializerSnapshot(CommonSerializerKeys.NAMESPACE_SERIALIZER); + + if (valueSnapshot == null) { + continue; + } + + result.add( + new StateSchemaInfo( + meta.getName(), + stateKind, + keySnapshot, + valueSnapshot, + mapKeySnapshot, + namespaceSnapshot)); + } + + return result; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaInfo.java new file mode 100644 index 00000000000..9031996dc8b --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/schema/StateSchemaInfo.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; + +import javax.annotation.Nullable; + +/** + * Carries the schema information extracted from a single keyed state entry in a savepoint, without + * requiring user POJO classes on the classpath. + * + * <p>Naming convention used throughout this package and {@code state.table}: "namespace" names the + * raw concept as Flink's runtime state backend knows it ({@link #namespaceSnapshot}), "window" the + * same concept once resolved to a user-facing table/column ({@link + * KeyedStateSchemaInfo.StateEntryInfo#windowLogicalType}). + */ +@Internal +public final class StateSchemaInfo { + + /** Name of the state as registered by the operator. */ + public final String stateName; + + /** The kind of state (VALUE, LIST, MAP, etc.). */ + public final StateDescriptor.Type stateKind; + + /** Serializer snapshot for the key type. */ + public final TypeSerializerSnapshot<?> keySnapshot; + + /** + * Serializer snapshot for the state value type. For MAP state this is the value type; use + * {@link #mapKeySnapshot} for the map key type. + */ + public final TypeSerializerSnapshot<?> valueSnapshot; + + /** + * Serializer snapshot for the map key type. Non-null only for {@link StateDescriptor.Type#MAP} + * state. + */ + @Nullable public final TypeSerializerSnapshot<?> mapKeySnapshot; + + /** + * Serializer snapshot for the state's namespace. Plain per-key state (the only kind the + * savepoint/checkpoint table connector can read) is registered with {@code VoidNamespace}; a + * different namespace (e.g. a window) means the state is scoped per-window rather than per-key, + * and cannot be exposed as a flat keyed table. + */ + @Nullable public final TypeSerializerSnapshot<?> namespaceSnapshot; + + public StateSchemaInfo( + String stateName, + StateDescriptor.Type stateKind, + TypeSerializerSnapshot<?> keySnapshot, + TypeSerializerSnapshot<?> valueSnapshot, + @Nullable TypeSerializerSnapshot<?> mapKeySnapshot, + @Nullable TypeSerializerSnapshot<?> namespaceSnapshot) { + this.stateName = stateName; + this.stateKind = stateKind; + this.keySnapshot = keySnapshot; + this.valueSnapshot = valueSnapshot; + this.mapKeySnapshot = mapKeySnapshot; + this.namespaceSnapshot = namespaceSnapshot; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/EmbeddedRocksDBKeyedStateReadingITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/EmbeddedRocksDBKeyedStateReadingITCase.java new file mode 100644 index 00000000000..3bfe8752f27 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/EmbeddedRocksDBKeyedStateReadingITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link KeyedStateReadingITCase} against the embedded RocksDB state backend. */ +public class EmbeddedRocksDBKeyedStateReadingITCase extends KeyedStateReadingITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "rocksdb"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/HashMapKeyedStateReadingITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/HashMapKeyedStateReadingITCase.java new file mode 100644 index 00000000000..bd9294ed210 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/HashMapKeyedStateReadingITCase.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.StateBackendOptions; + +/** Runs {@link KeyedStateReadingITCase} against the heap ({@code hashmap}) state backend. */ +public class HashMapKeyedStateReadingITCase extends KeyedStateReadingITCase { + + @Override + protected Configuration getConfiguration() { + return new Configuration().set(StateBackendOptions.STATE_BACKEND, "hashmap"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java new file mode 100644 index 00000000000..fa8b73340ea --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java @@ -0,0 +1,111 @@ +/* + * 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; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.state.api.runtime.SavepointLoader; +import org.apache.flink.state.api.schema.KeyedStateSchemaInfo; +import org.apache.flink.state.api.utils.SavepointTestBase; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Integration tests that write real keyed state through a MiniCluster job, take a savepoint at + * runtime, and read it back — verified against both the heap ({@code hashmap}) and RocksDB state + * backends (see {@code HashMapKeyedStateReadingITCase} / {@code + * EmbeddedRocksDBKeyedStateReadingITCase}) so that schema extraction and reads are checked against + * both keyed-state-handle formats. + */ +public abstract class KeyedStateReadingITCase extends SavepointTestBase { + + protected abstract Configuration getConfiguration(); + + // ------------------------------------------------------------------------- + // Schema extraction: RowData-typed internal SQL operator state + // ------------------------------------------------------------------------- + + @Test + public void testGroupAggAccStateSchemaExtraction() throws Exception { + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(getConfiguration()); + env.setParallelism(1); + + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + Tuple2<String, Long>[] data = + new Tuple2[] { + Tuple2.of("a", 1L), Tuple2.of("a", 2L), Tuple2.of("b", 3L), + }; + DataStream<Tuple2<String, Long>> source = env.addSource(createSource(data)); + tEnv.createTemporaryView( + "t", + source, + Schema.newBuilder().column("f0", "STRING").column("f1", "BIGINT").build()); + + Table result = + tEnv.sqlQuery( + "SELECT f0 AS `key`, COUNT(*) AS cnt, SUM(f1) AS total FROM t GROUP BY f0"); + + DataStream<Row> resultStream = tEnv.toChangelogStream(result); + resultStream.sinkTo(new DiscardingSink<>()); + + String savepointPath = takeSavepoint(env); + CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(savepointPath); + + // SQL-planned operators don't carry a user-assigned uid; find the aggregation operator by + // the state it registers. + OperatorIdentifier aggOpId = null; + for (OperatorIdentifier candidate : StateTableUtils.getOperatorIdentifiers(metadata)) { + List<String> stateNames = StateTableUtils.getKeyedStates(metadata, candidate); + if (stateNames.contains("accState")) { + aggOpId = candidate; + break; + } + } + assertNotNull(aggOpId, "Could not find operator with 'accState'"); + + KeyedStateSchemaInfo schemaInfo = StateTableUtils.getKeyedStateSchema(metadata, aggOpId); + KeyedStateSchemaInfo.StateEntryInfo accEntry = schemaInfo.stateSchemas.get("accState"); + assertNotNull(accEntry, "'accState' not found in extracted schema"); + + assertEquals(LogicalTypeRoot.ROW, accEntry.logicalType.getTypeRoot()); + RowType rowType = (RowType) accEntry.logicalType; + + // The accumulator row holds one field per aggregate call: COUNT(*) and SUM(f1). + assertEquals(2, rowType.getFieldCount()); + assertEquals(LogicalTypeRoot.BIGINT, rowType.getFields().get(0).getType().getTypeRoot()); + assertEquals(LogicalTypeRoot.BIGINT, rowType.getFields().get(1).getType().getTypeRoot()); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateTableUtilsTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateTableUtilsTest.java new file mode 100644 index 00000000000..34a6e16bda1 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/StateTableUtilsTest.java @@ -0,0 +1,129 @@ +/* + * 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; + +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.state.StateBackendLoader; +import org.apache.flink.state.api.runtime.SavepointLoader; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Unit tests for {@link StateTableUtils} that do not require a running Flink cluster. */ +public class StateTableUtilsTest { + + // ------------------------------------------------------------------------- + // getOperatorIdentifiers — filters operators without keyed state + // ------------------------------------------------------------------------- + + @Test + public void testGetOperatorIdentifiersFiltersEmptyOperators() { + OperatorState opState1 = new OperatorState(null, null, new OperatorID(1L, 2L), 1, 128); + OperatorState opState2 = new OperatorState(null, null, new OperatorID(3L, 4L), 2, 128); + + // No subtasks means no keyed state, regardless of how many such operators are present. + List<List<OperatorState>> cases = + Arrays.asList( + Collections.singletonList(opState1), + Collections.emptyList(), + Arrays.asList(opState1, opState2)); + + for (List<OperatorState> operators : cases) { + CheckpointMetadata metadata = + new CheckpointMetadata(1L, operators, Collections.emptyList()); + List<OperatorIdentifier> ids = StateTableUtils.getOperatorIdentifiers(metadata); + + assertNotNull(ids); + assertTrue( + ids.isEmpty(), + "Operators without keyed state should be filtered out, input: " + operators); + } + } + + // ------------------------------------------------------------------------- + // detectStateBackendType — reads real checkpoint metadata produced by other tests + // ------------------------------------------------------------------------- + + /** + * Checkpoint (native format) taken with the HashMap state backend, committed as a fixture for + * {@code StatefulJobSnapshotMigrationITCase} in flink-tests. Native-format keyed state handles + * retain their backend-specific type, so every keyed operator here must resolve to {@link + * StateBackendLoader#HASHMAP_STATE_BACKEND_NAME}. + */ + private static final String HASHMAP_CHECKPOINT_DIR = + "../../flink-tests/src/test/resources/" + + "new-stateful-udf-migration-itcase-flink1.20-hashmap-checkpoint"; + + /** + * Checkpoint (native format) taken with the RocksDB state backend, committed as a fixture for + * {@code StatefulJobSnapshotMigrationITCase} in flink-tests. Must resolve to {@link + * StateBackendLoader#ROCKSDB_STATE_BACKEND_NAME}. + */ + private static final String ROCKSDB_CHECKPOINT_DIR = + "../../flink-tests/src/test/resources/" + + "new-stateful-udf-migration-itcase-flink2.1-rocksdb-checkpoint"; + + @Test + public void testDetectStateBackendTypeFromHashMapCheckpoint() throws IOException { + assertAllKeyedOperatorsDetectAs( + HASHMAP_CHECKPOINT_DIR, StateBackendLoader.HASHMAP_STATE_BACKEND_NAME); + } + + @Test + public void testDetectStateBackendTypeFromRocksDBCheckpoint() throws IOException { + assertAllKeyedOperatorsDetectAs( + ROCKSDB_CHECKPOINT_DIR, StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME); + } + + /** + * Loads the checkpoint metadata at {@code checkpointDir} and asserts that every operator + * carrying keyed state resolves to exactly {@code expectedType}, and that at least one operator + * did so (i.e. the fixture actually exercises the detection logic). + */ + private static void assertAllKeyedOperatorsDetectAs(String checkpointDir, String expectedType) + throws IOException { + CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(checkpointDir); + assertFalse(metadata.getOperatorStates().isEmpty()); + + int operatorsWithKeyedState = 0; + for (OperatorState opState : metadata.getOperatorStates()) { + Optional<String> detected = StateTableUtils.detectStateBackendType(opState); + if (detected.isEmpty()) { + continue; + } + operatorsWithKeyedState++; + assertEquals(expectedType, detected.get(), "operator " + opState.getOperatorID()); + } + assertTrue( + operatorsWithKeyedState > 0, + "Expected at least one operator with keyed state in " + checkpointDir); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverterTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverterTest.java new file mode 100644 index 00000000000..3773b67dd67 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/SerializerSnapshotToLogicalTypeConverterTest.java @@ -0,0 +1,425 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.BooleanSerializer; +import org.apache.flink.api.common.typeutils.base.DoubleSerializer; +import org.apache.flink.api.common.typeutils.base.FloatSerializer; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.LongSerializer; +import org.apache.flink.api.common.typeutils.base.MapSerializer; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.typeutils.TypeExtractor; +import org.apache.flink.api.java.typeutils.runtime.NullableSerializer; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.formats.avro.typeutils.AvroTypeInfo; +import org.apache.flink.formats.avro.typeutils.GenericRecordAvroTypeInfo; +import org.apache.flink.table.runtime.typeutils.RowDataSerializer; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; + +import com.example.state.writer.job.schema.avro.AvroRecord; +import org.apache.avro.Schema; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; +import java.util.function.Supplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link SerializerSnapshotToLogicalTypeConverter}. */ +class SerializerSnapshotToLogicalTypeConverterTest { + + // ------------------------------------------------------------------------- + // POJO classes + // ------------------------------------------------------------------------- + + /** Simple POJO used to exercise field-by-field POJO schema extraction. */ + public static class SimplePojo { + public String name; + public int age; + public long score; + public boolean active; + } + + /** POJO with a nested POJO field, used to exercise recursive schema extraction. */ + public static class NestedPojo { + public String label; + public SimplePojo inner; + } + + // ------------------------------------------------------------------------- + // Primitive / scalar snapshots + // ------------------------------------------------------------------------- + + @Test + void testPrimitives() { + List<PrimitiveCase> cases = + Arrays.asList( + new PrimitiveCase( + IntSerializer.IntSerializerSnapshot::new, LogicalTypeRoot.INTEGER), + new PrimitiveCase( + LongSerializer.LongSerializerSnapshot::new, LogicalTypeRoot.BIGINT), + new PrimitiveCase( + FloatSerializer.FloatSerializerSnapshot::new, + LogicalTypeRoot.FLOAT), + new PrimitiveCase( + DoubleSerializer.DoubleSerializerSnapshot::new, + LogicalTypeRoot.DOUBLE), + new PrimitiveCase( + BooleanSerializer.BooleanSerializerSnapshot::new, + LogicalTypeRoot.BOOLEAN), + new PrimitiveCase( + StringSerializer.StringSerializerSnapshot::new, + LogicalTypeRoot.VARCHAR)); + + for (PrimitiveCase c : cases) { + var snapshot = c.snapshotSupplier.get(); + LogicalType t = convert(snapshot); + assertThat(t.getTypeRoot()) + .as("wrong type root for %s", snapshot.getClass().getSimpleName()) + .isEqualTo(c.expectedRoot); + } + + // Numeric primitives are non-nullable at the wire level; spot-check one representative. + assertThat(convert(new IntSerializer.IntSerializerSnapshot()).isNullable()).isFalse(); + } + + private static final class PrimitiveCase { + final Supplier<org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?>> + snapshotSupplier; + final LogicalTypeRoot expectedRoot; + + PrimitiveCase( + Supplier<org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?>> + snapshotSupplier, + LogicalTypeRoot expectedRoot) { + this.snapshotSupplier = snapshotSupplier; + this.expectedRoot = expectedRoot; + } + } + + // ------------------------------------------------------------------------- + // Composite types + // ------------------------------------------------------------------------- + + @Test + void testListOfString() { + ListSerializer<String> ser = new ListSerializer<>(StringSerializer.INSTANCE); + LogicalType t = convert(ser.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ARRAY); + ArrayType at = (ArrayType) t; + assertThat(at.getElementType().getTypeRoot()).isEqualTo(LogicalTypeRoot.VARCHAR); + } + + @Test + void testMapStringToLong() { + MapSerializer<String, Long> ser = + new MapSerializer<>(StringSerializer.INSTANCE, LongSerializer.INSTANCE); + LogicalType t = convert(ser.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.MAP); + MapType mt = (MapType) t; + assertThat(mt.getKeyType().getTypeRoot()).isEqualTo(LogicalTypeRoot.VARCHAR); + assertThat(mt.getValueType().getTypeRoot()).isEqualTo(LogicalTypeRoot.BIGINT); + } + + @Test + void testNullableWrapsNestedTypeAsNullable() { + // LongSerializer alone always maps to a non-nullable BIGINT (see testPrimitives); wrapping + // it in NullableSerializer must flip only the nullability, not the underlying type. + TypeSerializer<Long> wrapped = NullableSerializer.wrap(LongSerializer.INSTANCE, true); + LogicalType t = convert(wrapped.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.BIGINT); + assertThat(t.isNullable()).isTrue(); + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + void testTuple() { + TupleSerializer<Tuple2<Integer, String>> ser = + new TupleSerializer<>( + (Class) Tuple2.class, + new TypeSerializer<?>[] { + IntSerializer.INSTANCE, StringSerializer.INSTANCE + }); + LogicalType t = convert(ser.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + // Tuples have no field names in the serializer snapshot, so fields fall back to + // positional names, same as the RowData-without-field-names case. + assertField(rt, "f0", LogicalTypeRoot.INTEGER); + assertField(rt, "f1", LogicalTypeRoot.VARCHAR); + } + + // ------------------------------------------------------------------------- + // POJO types + // ------------------------------------------------------------------------- + + @Test + void testSimplePojo() { + LogicalType t = convertPojoType(SimplePojo.class); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(4); + + assertField(rt, "name", LogicalTypeRoot.VARCHAR); + assertField(rt, "age", LogicalTypeRoot.INTEGER); + assertField(rt, "score", LogicalTypeRoot.BIGINT); + assertField(rt, "active", LogicalTypeRoot.BOOLEAN); + } + + @Test + void testNestedPojo() { + LogicalType t = convertPojoType(NestedPojo.class); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + assertField(rt, "label", LogicalTypeRoot.VARCHAR); + + // The nested 'inner' field should map to ROW + RowType.RowField innerField = findField(rt, "inner"); + assertThat(innerField).isNotNull(); + assertThat(innerField.getType().getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType innerRow = (RowType) innerField.getType(); + assertField(innerRow, "name", LogicalTypeRoot.VARCHAR); + assertField(innerRow, "age", LogicalTypeRoot.INTEGER); + } + + @Test + void testPojoFieldNamesPreservedWithoutClass() { + // Even without the POJO class on the classpath, field names should be available. + var snapshot = buildPojoSnapshot(SimplePojo.class); + // Field name extraction happens via the snapshot, no class needed + LogicalType t = SerializerSnapshotToLogicalTypeConverter.convert(snapshot); + RowType rt = (RowType) t; + List<String> names = rt.getFieldNames(); + assertThat(names).contains("name", "age", "score", "active"); + } + + // ------------------------------------------------------------------------- + // Avro types + // ------------------------------------------------------------------------- + + @Test + void testAvroSpecificRecord() { + // AvroRecord has one field: longData (long) + LogicalType t = convertAvroSpecificType(AvroRecord.class); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(1); + assertField(rt, "longData", LogicalTypeRoot.BIGINT); + } + + @Test + void testAvroGenericRecord() { + // convertAvro() reads only the embedded writer Schema (see + // SerializerSnapshotToLogicalTypeConverter#convertAvro), so this also covers the + // "specific record class missing at read time" fallback: AvroSerializerSnapshot degrades + // to GenericRecord.class in that case, but the schema-derived RowType is identical either + // way. The actual missing-class read path is covered end-to-end by + // StateCatalogGeneratedSavepointITCase#testReadAvroKeyedStateFromSchemaDiscovery. + Schema schema = AvroRecord.getClassSchema(); + LogicalType t = convertAvroGenericType(schema); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(1); + assertField(rt, "longData", LogicalTypeRoot.BIGINT); + } + + // ------------------------------------------------------------------------- + // RowData types + // ------------------------------------------------------------------------- + + @Test + void testRowDataWithFieldNames() { + RowType rowType = + RowType.of( + new LogicalType[] {new IntType(), VarCharType.STRING_TYPE}, + new String[] {"id", "name"}); + RowDataSerializer serializer = new RowDataSerializer(rowType); + + LogicalType t = convert(serializer.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + assertField(rt, "id", LogicalTypeRoot.INTEGER); + assertField(rt, "name", LogicalTypeRoot.VARCHAR); + } + + @Test + void testRowDataWithoutFieldNamesFallsBackToPositional() { + // Many production call sites (e.g. window operators) build a RowDataSerializer from a + // bare LogicalType[], so no field names are available. The converter must still produce + // a usable RowType, falling back to positional names like the Tuple case. + RowDataSerializer serializer = new RowDataSerializer(new IntType(), new BigIntType()); + + LogicalType t = convert(serializer.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + assertField(rt, "f0", LogicalTypeRoot.INTEGER); + assertField(rt, "f1", LogicalTypeRoot.BIGINT); + } + + @Test + void testNestedRowData() { + RowType innerType = + RowType.of( + new LogicalType[] {new IntType(), VarCharType.STRING_TYPE}, + new String[] {"innerId", "innerName"}); + RowType outerType = + RowType.of( + new LogicalType[] {VarCharType.STRING_TYPE, innerType}, + new String[] {"label", "inner"}); + RowDataSerializer serializer = new RowDataSerializer(outerType); + + LogicalType t = convert(serializer.snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + assertField(rt, "label", LogicalTypeRoot.VARCHAR); + + RowType.RowField innerField = findField(rt, "inner"); + assertThat(innerField).isNotNull(); + assertThat(innerField.getType().getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + RowType innerRow = (RowType) innerField.getType(); + assertField(innerRow, "innerId", LogicalTypeRoot.INTEGER); + assertField(innerRow, "innerName", LogicalTypeRoot.VARCHAR); + } + + // ------------------------------------------------------------------------- + // Window namespace types + // ------------------------------------------------------------------------- + + @Test + void testTimeWindow() { + LogicalType t = + convert( + new org.apache.flink.streaming.api.windowing.windows.TimeWindow.Serializer() + .snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + assertThat(t.isNullable()).isFalse(); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(2); + assertField(rt, "window_start", LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE); + assertField(rt, "window_end", LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE); + } + + @Test + void testGlobalWindow() { + LogicalType t = + convert( + new org.apache.flink.streaming.api.windowing.windows.GlobalWindow + .Serializer() + .snapshotConfiguration()); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.ROW); + assertThat(t.isNullable()).isFalse(); + RowType rt = (RowType) t; + assertThat(rt.getFieldCount()).isEqualTo(0); + } + + // ------------------------------------------------------------------------- + // Null / unknown snapshot + // ------------------------------------------------------------------------- + + @Test + void testNullSnapshot() { + LogicalType t = SerializerSnapshotToLogicalTypeConverter.convert(null); + assertThat(t.getTypeRoot()).isEqualTo(LogicalTypeRoot.VARBINARY); + } + + @Test + void testVoidNamespaceSnapshotUnsupported() { + // VoidNamespace is filtered out by StateTableUtils before ever reaching the converter + // (plain per-key state has no namespace to convert); confirm it stays unsupported here. + assertThatThrownBy( + () -> + convert( + new org.apache.flink.runtime.state.VoidNamespaceSerializer() + .snapshotConfiguration())) + .isInstanceOf(UnsupportedOperationException.class); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static LogicalType convert( + org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?> snapshot) { + return SerializerSnapshotToLogicalTypeConverter.convert(snapshot); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static LogicalType convertPojoType(Class<?> pojoClass) { + var snapshot = buildPojoSnapshot(pojoClass); + return SerializerSnapshotToLogicalTypeConverter.convert(snapshot); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?> + buildPojoSnapshot(Class<?> clazz) { + return TypeExtractor.createTypeInfo(clazz) + .createSerializer(new SerializerConfigImpl()) + .snapshotConfiguration(); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static <T extends org.apache.avro.specific.SpecificRecordBase> + LogicalType convertAvroSpecificType(Class<T> avroClass) { + return SerializerSnapshotToLogicalTypeConverter.convert( + new AvroTypeInfo<>(avroClass) + .createSerializer(new SerializerConfigImpl()) + .snapshotConfiguration()); + } + + private static LogicalType convertAvroGenericType(Schema schema) { + return SerializerSnapshotToLogicalTypeConverter.convert( + new GenericRecordAvroTypeInfo(schema) + .createSerializer(new SerializerConfigImpl()) + .snapshotConfiguration()); + } + + private static void assertField(RowType row, String name, LogicalTypeRoot expectedRoot) { + RowType.RowField field = findField(row, name); + assertThat(field).as("Field '%s' not found in row type", name).isNotNull(); + assertThat(field.getType().getTypeRoot()) + .as("Wrong type for field '%s'", name) + .isEqualTo(expectedRoot); + } + + private static RowType.RowField findField(RowType row, String name) { + return row.getFields().stream() + .filter(f -> f.getName().equals(name)) + .findFirst() + .orElse(null); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateSchemaExtractorTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateSchemaExtractorTest.java new file mode 100644 index 00000000000..7849dcdd672 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/schema/StateSchemaExtractorTest.java @@ -0,0 +1,204 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.api.schema; + +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.api.common.typeutils.base.LongSerializer; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.runtime.state.KeyedBackendSerializationProxy; +import org.apache.flink.runtime.state.VoidNamespaceSerializer; +import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link StateSchemaExtractor}. */ +class StateSchemaExtractorTest { + + // ------------------------------------------------------------------------- + // Tests + // ------------------------------------------------------------------------- + + @Test + void testExtractSingleValueState() throws IOException { + StateMetaInfoSnapshot stateSnap = + buildValueStateSnapshot("my-state", StateDescriptor.Type.VALUE); + + KeyedBackendSerializationProxy<Integer> proxy = + new KeyedBackendSerializationProxy<>( + IntSerializer.INSTANCE, Collections.singletonList(stateSnap), false); + + List<StateSchemaInfo> result = roundTrip(proxy); + + assertThat(result).hasSize(1); + StateSchemaInfo info = result.get(0); + assertThat(info.stateName).isEqualTo("my-state"); + assertThat(info.stateKind).isEqualTo(StateDescriptor.Type.VALUE); + assertThat(info.valueSnapshot).isInstanceOf(IntSerializer.IntSerializerSnapshot.class); + assertThat(info.keySnapshot).isInstanceOf(IntSerializer.IntSerializerSnapshot.class); + assertThat(info.mapKeySnapshot).isNull(); + } + + @Test + void testExtractMultipleStates() throws IOException { + StateMetaInfoSnapshot valueState = + buildValueStateSnapshot("int-state", StateDescriptor.Type.VALUE); + StateMetaInfoSnapshot stringState = + buildValueStateSnapshotWithStringValue("str-state", StateDescriptor.Type.VALUE); + + KeyedBackendSerializationProxy<Integer> proxy = + new KeyedBackendSerializationProxy<>( + IntSerializer.INSTANCE, Arrays.asList(valueState, stringState), false); + + List<StateSchemaInfo> result = roundTrip(proxy); + + assertThat(result).hasSize(2); + + assertThat(result.get(0).stateName).isEqualTo("int-state"); + assertThat(result.get(0).valueSnapshot) + .isInstanceOf(IntSerializer.IntSerializerSnapshot.class); + + assertThat(result.get(1).stateName).isEqualTo("str-state"); + assertThat(result.get(1).valueSnapshot) + .isInstanceOf(StringSerializer.StringSerializerSnapshot.class); + } + + @Test + void testExtractMapState() throws IOException { + Map<String, String> options = new HashMap<>(); + options.put( + StateMetaInfoSnapshot.CommonOptionsKeys.KEYED_STATE_TYPE.toString(), + StateDescriptor.Type.MAP.toString()); + + Map<String, org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?>> + serializerSnapshots = new LinkedHashMap<>(); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString(), + new VoidNamespaceSerializer.VoidNamespaceSerializerSnapshot()); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.USER_KEY_SERIALIZER.toString(), + new StringSerializer.StringSerializerSnapshot()); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(), + new LongSerializer.LongSerializerSnapshot()); + + StateMetaInfoSnapshot mapSnap = + new StateMetaInfoSnapshot( + "map-state", + StateMetaInfoSnapshot.BackendStateType.KEY_VALUE, + options, + serializerSnapshots); + + KeyedBackendSerializationProxy<Integer> proxy = + new KeyedBackendSerializationProxy<>( + IntSerializer.INSTANCE, Collections.singletonList(mapSnap), false); + + List<StateSchemaInfo> result = roundTrip(proxy); + + assertThat(result).hasSize(1); + StateSchemaInfo info = result.get(0); + assertThat(info.stateName).isEqualTo("map-state"); + assertThat(info.stateKind).isEqualTo(StateDescriptor.Type.MAP); + assertThat(info.mapKeySnapshot) + .isInstanceOf(StringSerializer.StringSerializerSnapshot.class); + assertThat(info.valueSnapshot).isInstanceOf(LongSerializer.LongSerializerSnapshot.class); + } + + @Test + void testEmptyStates() throws IOException { + KeyedBackendSerializationProxy<Integer> proxy = + new KeyedBackendSerializationProxy<>( + IntSerializer.INSTANCE, Collections.emptyList(), false); + + List<StateSchemaInfo> result = roundTrip(proxy); + + assertThat(result).isNotNull().isEmpty(); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + private static StateMetaInfoSnapshot buildValueStateSnapshot( + String name, StateDescriptor.Type type) { + Map<String, String> options = new HashMap<>(); + options.put( + StateMetaInfoSnapshot.CommonOptionsKeys.KEYED_STATE_TYPE.toString(), + type.toString()); + + Map<String, org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?>> + serializerSnapshots = new LinkedHashMap<>(); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString(), + new VoidNamespaceSerializer.VoidNamespaceSerializerSnapshot()); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(), + new IntSerializer.IntSerializerSnapshot()); + + return new StateMetaInfoSnapshot( + name, + StateMetaInfoSnapshot.BackendStateType.KEY_VALUE, + options, + serializerSnapshots); + } + + private static StateMetaInfoSnapshot buildValueStateSnapshotWithStringValue( + String name, StateDescriptor.Type type) { + Map<String, String> options = new HashMap<>(); + options.put( + StateMetaInfoSnapshot.CommonOptionsKeys.KEYED_STATE_TYPE.toString(), + type.toString()); + + Map<String, org.apache.flink.api.common.typeutils.TypeSerializerSnapshot<?>> + serializerSnapshots = new LinkedHashMap<>(); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.NAMESPACE_SERIALIZER.toString(), + new VoidNamespaceSerializer.VoidNamespaceSerializerSnapshot()); + serializerSnapshots.put( + StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(), + new StringSerializer.StringSerializerSnapshot()); + + return new StateMetaInfoSnapshot( + name, + StateMetaInfoSnapshot.BackendStateType.KEY_VALUE, + options, + serializerSnapshots); + } + + private static List<StateSchemaInfo> roundTrip(KeyedBackendSerializationProxy<?> proxy) + throws IOException { + DataOutputSerializer out = new DataOutputSerializer(256); + proxy.write(out); + + DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer()); + return StateSchemaExtractor.extractSchema(in); + } +}
