gaborgsomogyi commented on code in PR #28837:
URL: https://github.com/apache/flink/pull/28837#discussion_r3767600900
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java:
##########
@@ -206,7 +212,17 @@ void writeSnapshotData(DataOutputView out) throws
IOException {
private static <T> PojoSerializerSnapshotData<T> readSnapshotData(
DataInputView in, ClassLoader userCodeClassLoader) throws
IOException {
- Class<T> pojoClass = InstantiationUtil.resolveClassByName(in,
userCodeClassLoader);
+ final String pojoClassName = in.readUTF();
+ Class<T> pojoClass = null;
Review Comment:
This is heavier than a missing-class fix: it silently downgrades the
guarantee for the whole compatibility check, not just this one case.
Before this PR, a missing class was the one kind of incompatibility caught
eagerly and unconditionally, at parse time, independent of whether the state
was ever accessed. Every *other* kind of incompatibility (field/type changes on
a class that does resolve) has always relied solely on
`resolveSchemaCompatibility()`, which only runs if `getState()` is called for
that state. For RocksDB, that's a pre-existing characteristic of the whole
compatibility mechanism, not something new here.
This PR removes missing-class's special eager check, so it now falls back to
that same access-gated mechanism too. Practically: on RocksDB, a state that a
job upgrade stops referencing is never checked again for *any* form of
incompatibility, missing class included, not because this PR weakens RocksDB's
compatibility check in general, but because it removes the one thing that used
to be exempt from that pre-existing weakness.
Operationally this matters because restoring a new job version against a
prod savepoint in staging/CI is only a reliable schema-compatibility gate if
it's unconditional. Whether it still catches a missing class now depends on
which states happen to get touched during that run. It can pass staging clean
and fail in production later with no trace back to the deploy that caused it.
Could this be gated on whether a `CustomRestoreSerializerFactory` is already
registered at parse time, falling back to the original unconditional throw
otherwise? That keeps missing-class detection's original exemption intact for
ordinary restores while still enabling this feature's classless read.
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoDeserializerCompatibilitySnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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.api.java.typeutils.runtime;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import javax.annotation.Nullable;
+
+/**
+ * A {@link TypeSerializerSnapshot} for deserializers that can read POJO
binary data without the
+ * user POJO class being on the classpath, such as the {@code
PojoToRowDataDeserializer} of the
+ * State Processing API. It declares itself {@link
+ * TypeSerializerSchemaCompatibility#compatibleAsIs() compatible as-is} with
any stored {@link
+ * PojoSerializerSnapshot}.
+ *
+ * <p>The snapshot only ever exists in memory, wrapping the live deserializer
it was created from:
+ * composite compatibility checks (e.g. {@code
+ * CompositeTypeSerializerSnapshot#resolveOuterSchemaCompatibility}) restore
the "new" side of a
+ * composite serializer even when nested-level compatibility already
short-circuited to {@code
+ * compatibleAsIs()}, so {@link #restoreSerializer()} hands back the wrapped
instance rather than
+ * reconstructing one from persisted bytes.
+ */
+@Internal
+public final class PojoDeserializerCompatibilitySnapshot<T> implements
TypeSerializerSnapshot<T> {
Review Comment:
Does this need to live in core? Only consumer named in its own javadoc is
PojoToRowDataDeserializer in the State Processing API, and it doesn't need
core-internal access, just a public instanceof check. Could this move next to
its consumer instead?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input.deserializer;
+
+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.api.java.typeutils.runtime.PojoDeserializerCompatibilitySnapshot;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import
org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A {@link TypeSerializer} that reads the POJO binary format written by {@link
+ * org.apache.flink.api.java.typeutils.runtime.PojoSerializer} and produces
{@link GenericRowData}.
+ *
+ * <p>This deserializer does <em>not</em> require the user POJO class to be on
the classpath. It
+ * mirrors the exact binary protocol of {@code PojoSerializer}:
+ *
+ * <pre>{@code
+ * 1 byte: flags (bitmask)
+ * 0x01 IS_NULL → value is null, return null
+ * 0x02 NO_SUBCLASS → exact POJO class: read numFields × (isNull
boolean + field bytes)
+ * 0x08 IS_TAGGED_SUBCLASS → 1 byte subclass tag; delegate to registered
subclass deserializer
+ * 0x04 IS_SUBCLASS → UTF class name (must be read); Kryo not
supported → throws IOException
+ * }</pre>
+ *
+ * <p>Use {@link #create(PojoSerializerSnapshot)} to build an instance from a
savepoint snapshot.
+ */
+@Internal
+public final class PojoToRowDataDeserializer extends TypeSerializer<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PojoToRowDataDeserializer.class);
+
+ // Mirrors constants in PojoSerializer
+ static final int IS_NULL = 0x01;
+ static final int NO_SUBCLASS = 0x02;
+ static final int IS_SUBCLASS = 0x04;
+ static final int IS_TAGGED_SUBCLASS = 0x08;
+
+ private final int numFields;
+ private final TypeSerializer<?>[] fieldDeserializers;
+ private final LogicalType[] fieldTypes;
+ private final String[] fieldNames;
+ private final List<PojoToRowDataDeserializer>
registeredSubclassDeserializers;
+
+ /**
+ * Builds a {@link PojoToRowDataDeserializer} from a {@link
PojoSerializerSnapshot}.
+ *
+ * <p>For each field:
+ *
+ * <ul>
+ * <li>If the field snapshot is itself a {@link PojoSerializerSnapshot},
this method recurses
+ * to build a nested {@link PojoToRowDataDeserializer}.
+ * <li>For all other field types, the field's original serializer is
restored via {@link
+ * TypeSerializerSnapshot#restoreSerializer()}.
+ * </ul>
+ *
+ * <p>Registered subclasses are handled by building a deserializer for
each registered subclass
+ * snapshot in order (matching the tag index used in the binary format).
+ *
+ * @throws IllegalStateException if a required field serializer snapshot
is absent
+ */
+ public static PojoToRowDataDeserializer create(PojoSerializerSnapshot<?>
snapshot) {
+ List<AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>>
fieldEntries =
+ snapshot.getFieldSnapshotEntries();
+
+ List<TypeSerializer<?>> fieldDeserializerList = new
ArrayList<>(fieldEntries.size());
+ List<LogicalType> fieldTypeList = new ArrayList<>(fieldEntries.size());
+ List<String> fieldNameList = new ArrayList<>(fieldEntries.size());
+
+ for (AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>> entry
: fieldEntries) {
+ String fieldName = entry.getKey();
+ TypeSerializerSnapshot<?> fieldSnapshot = entry.getValue();
+
+ if (fieldSnapshot == null) {
+ throw new IllegalStateException(
+ "Cannot build deserializer for field '"
+ + fieldName
+ + "': its serializer snapshot was not readable
from the savepoint. "
+ + "This field cannot be deserialized without
the original snapshot.");
+ }
+
+ TypeSerializer<?> fieldDeserializer;
+ if (fieldSnapshot instanceof PojoSerializerSnapshot) {
+ fieldDeserializer = create((PojoSerializerSnapshot<?>)
fieldSnapshot);
+ } else {
+ fieldDeserializer = fieldSnapshot.restoreSerializer();
+ }
+
+ fieldDeserializerList.add(fieldDeserializer);
+
fieldTypeList.add(SerializerSnapshotToLogicalTypeConverter.convert(fieldSnapshot));
+ fieldNameList.add(fieldName);
+ }
+
+ List<TypeSerializerSnapshot<?>> subSnapshots =
+ snapshot.getRegisteredSubclassSnapshotsOrdered();
+ List<PojoToRowDataDeserializer> subDeserializers = new
ArrayList<>(subSnapshots.size());
+ for (TypeSerializerSnapshot<?> subSnap : subSnapshots) {
+ subDeserializers.add(
+ subSnap instanceof PojoSerializerSnapshot
+ ? create((PojoSerializerSnapshot<?>) subSnap)
+ : null);
+ }
+
+ return new PojoToRowDataDeserializer(
+ fieldDeserializerList.toArray(new TypeSerializer[0]),
+ fieldTypeList.toArray(new LogicalType[0]),
+ fieldNameList.toArray(new String[0]),
+ subDeserializers);
+ }
+
+ PojoToRowDataDeserializer(
+ TypeSerializer<?>[] fieldDeserializers,
+ LogicalType[] fieldTypes,
+ String[] fieldNames,
+ List<PojoToRowDataDeserializer> registeredSubclassDeserializers) {
+ this.numFields = fieldDeserializers.length;
+ this.fieldDeserializers = fieldDeserializers;
+ this.fieldTypes = fieldTypes;
+ this.fieldNames = fieldNames;
+ this.registeredSubclassDeserializers = registeredSubclassDeserializers;
+ }
+
+ @Override
+ public RowData deserialize(DataInputView source) throws IOException {
+ int flags = source.readByte() & 0xFF;
+
+ if ((flags & IS_NULL) != 0) {
+ return null;
+ }
+
+ if ((flags & NO_SUBCLASS) != 0) {
+ return readFields(source);
+ }
+
+ if ((flags & IS_TAGGED_SUBCLASS) != 0) {
+ int tag = source.readByte() & 0xFF;
+ if (tag < registeredSubclassDeserializers.size()) {
+ PojoToRowDataDeserializer subDeserializer =
+ registeredSubclassDeserializers.get(tag);
+ if (subDeserializer == null) {
+ // Either the subclass's own snapshot was unreadable, or
the subclass is not
+ // itself a POJO (e.g. it falls back to Kryo) — either way
we have no way to
+ // decode its bytes, and, like the IS_SUBCLASS/Kryo case
below, its length is
+ // unknown so the bytes cannot even be skipped.
+ throw new IOException(
+ "Cannot deserialize registered POJO subclass at
tag "
+ + tag
+ + ": its serializer snapshot is missing or
is not a POJO "
+ + "serializer (e.g. it uses Kryo), which
requires the class on "
+ + "the classpath.");
+ }
+ return subDeserializer.deserialize(source);
+ }
+ throw new IOException(
+ "Unknown registered subclass tag "
+ + tag
+ + " (have "
+ + registeredSubclassDeserializers.size()
+ + " registered). The savepoint may have been
written with more subclasses registered.");
+ }
+
+ if ((flags & IS_SUBCLASS) != 0) {
+ String className = source.readUTF();
+ throw new IOException(
+ "Cannot deserialize POJO subclass '"
+ + className
+ + "': the subclass uses Kryo serialization, which
requires the class on the"
+ + " classpath. Kryo-encoded bytes have unknown
length and cannot be skipped."
+ + " Register the subclass or add the JAR to the
classpath.");
+ }
+
+ throw new IOException("Unrecognised POJO flags byte: 0x" +
Integer.toHexString(flags));
+ }
+
+ @Override
+ public RowData deserialize(RowData reuse, DataInputView source) throws
IOException {
+ return deserialize(source);
+ }
+
+ private GenericRowData readFields(DataInputView source) throws IOException
{
+ GenericRowData row = new GenericRowData(numFields);
+ for (int i = 0; i < numFields; i++) {
+ boolean isNull = source.readBoolean();
+ if (isNull) {
+ row.setField(i, null);
+ continue;
+ }
+ Object raw;
+ try {
+ raw = fieldDeserializers[i].deserialize(source);
+ } catch (Exception e) {
Review Comment:
PojoSerializer.deserialize(), the reference implementation this mirrors,
does not catch per-field failures at all, it fails loudly. This
catch-and-null-with-warn pattern recurs in several other places across this PR
(StateTableUtils, SavepointTypeInfoResolver, KeyedStateInputFormat, etc). Why
is it needed here specifically? A real bug becomes indistinguishable from
expected messy data, silently, in query results.
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverter.java:
##########
@@ -0,0 +1,288 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input.deserializer;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.DecimalType;
+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.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+
+import javax.annotation.Nullable;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.StreamSupport;
+
+/**
+ * Converts external Java objects (as produced by DataStream serializers) to
Flink table internal
+ * types as expected by {@link GenericRowData}.
+ *
+ * <p>Conversion rules:
+ *
+ * <ul>
+ * <li>{@link String} → {@link StringData}
+ * <li>{@link BigDecimal} → {@link DecimalData} (precision/scale from {@link
DecimalType})
+ * <li>{@link ByteBuffer} or {@code byte[]} → {@link DecimalData} (unscaled
bytes)
+ * <li>{@link ByteBuffer} → {@code byte[]} for BINARY/VARBINARY
+ * <li>{@link java.sql.Date}, {@link LocalDate} → {@code int} (days since
epoch)
+ * <li>{@link Timestamp}, {@link Instant}, {@link LocalDateTime} → {@link
TimestampData}
+ * <li>{@link List}, arrays, {@link Iterable} → {@link GenericArrayData}
(elements recursively
+ * converted)
+ * <li>{@link Map} or {@link Iterable} of {@link Map.Entry} → {@link
GenericMapData} (keys/values
+ * recursively converted)
+ * <li>{@link Row} → {@link GenericRowData} (fields recursively converted)
+ * <li>{@link RowData} subtypes → passed through unchanged
+ * <li>Primitives (boxed) → passed through unchanged
+ * </ul>
+ */
+@Internal
+public final class InternalTypeConverter {
+
+ private InternalTypeConverter() {}
+
+ /**
+ * Converts {@code value} to the Flink table internal representation
dictated by {@code type}.
+ *
+ * @param value the raw Java object; may be null
+ * @param type the target logical type; used to drive nested conversions
+ * @return the converted value, or null if value is null
+ */
+ @Nullable
+ public static Object toInternal(@Nullable Object value, LogicalType type) {
+ if (value == null) {
+ return null;
+ }
+
+ switch (type.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ if (value instanceof StringData) {
+ return value;
+ }
+ return StringData.fromString(value.toString());
+
+ case BOOLEAN:
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ case FLOAT:
+ case DOUBLE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case INTERVAL_YEAR_MONTH:
+ case INTERVAL_DAY_TIME:
+ return value;
+
+ case DECIMAL:
+ if (value instanceof DecimalData) {
+ return value;
+ }
+ if (value instanceof BigDecimal) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromBigDecimal(
+ (BigDecimal) value, dt.getPrecision(),
dt.getScale());
+ }
+ if (value instanceof ByteBuffer) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromUnscaledBytes(
+ toByteArray((ByteBuffer) value),
dt.getPrecision(), dt.getScale());
+ }
+ if (value instanceof byte[]) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromUnscaledBytes(
+ (byte[]) value, dt.getPrecision(), dt.getScale());
+ }
+ return value;
+
+ case DATE:
+ if (value instanceof Integer) {
+ return value;
+ }
+ if (value instanceof java.sql.Date) {
+ return (int) ((java.sql.Date)
value).toLocalDate().toEpochDay();
+ }
+ if (value instanceof LocalDate) {
+ return (int) ((LocalDate) value).toEpochDay();
+ }
+ return value;
+
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ if (value instanceof TimestampData) {
+ return value;
+ }
+ if (value instanceof Timestamp) {
+ return TimestampData.fromTimestamp((Timestamp) value);
+ }
+ if (value instanceof Instant) {
+ return TimestampData.fromInstant((Instant) value);
+ }
+ if (value instanceof LocalDateTime) {
+ return TimestampData.fromLocalDateTime((LocalDateTime)
value);
+ }
+ return value;
+
+ case BINARY:
+ case VARBINARY:
+ if (value instanceof ByteBuffer) {
+ return toByteArray((ByteBuffer) value);
+ }
+ return value;
+
+ case NULL:
+ return null;
+
+ case ROW:
+ case STRUCTURED_TYPE:
+ if (value instanceof GenericRowData) {
+ return value;
+ }
+ if (value instanceof Row) {
+ return rowToGenericRowData((Row) value, (RowType) type);
+ }
+ return value;
+
+ case ARRAY:
+ if (value instanceof GenericArrayData) {
+ return value;
+ }
+ ArrayType at = (ArrayType) type;
+ if (value instanceof Object[]) {
+ return objectArrayToArrayData((Object[]) value,
at.getElementType());
+ }
+ if (value instanceof Iterable) {
+ return iterableToArrayData((Iterable<?>) value,
at.getElementType());
+ }
+ return value;
+
+ case MAP:
+ if (value instanceof GenericMapData) {
+ return value;
+ }
+ MapType mt = (MapType) type;
+ if (value instanceof Map) {
+ return mapToMapData((Map<?, ?>) value, mt.getKeyType(),
mt.getValueType());
+ }
+ if (value instanceof Iterable) {
+ return mapEntryIterableToMapData(
+ (Iterable<?>) value, mt.getKeyType(),
mt.getValueType());
+ }
+ return value;
+
+ case MULTISET:
+ // MultisetType is not a MapType: it has only an element type,
represented
+ // internally as Map<element, Integer> (element ->
multiplicity).
+ if (value instanceof GenericMapData) {
+ return value;
+ }
+ LogicalType elementType = ((MultisetType)
type).getElementType();
+ if (value instanceof Map) {
+ return mapToMapData((Map<?, ?>) value, elementType, new
IntType());
+ }
+ if (value instanceof Iterable) {
+ return mapEntryIterableToMapData(
+ (Iterable<?>) value, elementType, new IntType());
+ }
+ return value;
+
+ default:
+ return value;
Review Comment:
Minor: `default: return value;` here silently passes through unhandled
`LogicalTypeRoot`s, unlike every other enum dispatch in this PR, which throws
`UnsupportedOperationException` on unhandled cases. Is this different somehow?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java:
##########
@@ -0,0 +1,641 @@
+/*
+ * 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.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.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.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) {
Review Comment:
extractSchema()'s own contract already returns an empty list for no keyed
state, no exception involved. So anything landing here is either the documented
IOException (corrupted header) or a bug, and both silently disappear this
operator from the catalog listing with only a WARN log. What is the
plan/justification for hiding these from SHOW TABLES instead of surfacing them?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointDynamicTableSourceFactory.java:
##########
@@ -89,207 +55,152 @@ public DynamicTableSource
createDynamicTableSource(Context context) {
final String statePath = options.get(STATE_PATH);
final OperatorIdentifier operatorIdentifier =
getOperatorIdentifier(options);
- final Map<String, StateMetaInfoSnapshot> preloadedStateMetadata =
- preloadStateMetadata(statePath, operatorIdentifier);
-
- // Create resolver with preloaded metadata
- SavepointTypeInfoResolver typeResolver =
- new SavepointTypeInfoResolver(preloadedStateMetadata,
serializerConfig);
-
- final Tuple2<Integer, int[]> keyValueProjections =
- createKeyValueProjections(context.getCatalogTable());
+ SavepointConnectorOptions.StateReaderMode readerMode =
options.get(STATE_READER_MODE);
+ switch (readerMode) {
+ case KEYED:
+ return createKeyedDynamicTableSource(
+ context,
+ options,
+ serializerConfig,
+ stateBackendType,
+ statePath,
+ operatorIdentifier);
+ case KEYED_FLAT:
+ return createFlattenedDynamicTableSource(
+ context,
+ options,
+ serializerConfig,
+ stateBackendType,
+ statePath,
+ operatorIdentifier);
+ default:
+ throw new IllegalArgumentException("Unsupported state reader
mode: " + readerMode);
+ }
+ }
- LogicalType logicalType =
context.getPhysicalRowDataType().getLogicalType();
- Preconditions.checkArgument(logicalType.is(LogicalTypeRoot.ROW), "Row
data type expected.");
- RowType rowType = (RowType) logicalType;
+ /**
+ * Creates a {@link SavepointDynamicTableSource} for the general
keyed-state table (selected via
+ * {@link SavepointConnectorOptions#STATE_READER_MODE} being set to {@link
+ * SavepointConnectorOptions.StateReaderMode#KEYED}, the default).
+ */
+ private DynamicTableSource createKeyedDynamicTableSource(
+ Context context,
+ Configuration options,
+ SerializerConfig serializerConfig,
+ String stateBackendType,
+ String statePath,
+ OperatorIdentifier operatorIdentifier) {
Set<ConfigOption<?>> requiredOptions = new
HashSet<>(requiredOptions());
Set<ConfigOption<?>> optionalOptions = new
HashSet<>(optionalOptions());
- RowType.RowField keyRowField =
rowType.getFields().get(keyValueProjections.f0);
- ConfigOption<String> keyFormatOption =
- optionOf(keyRowField.getName(),
VALUE_CLASS).stringType().noDefaultValue();
- optionalOptions.add(keyFormatOption);
-
- ConfigOption<String> keyTypeInfoFactoryOption =
- optionOf(keyRowField.getName(),
VALUE_TYPE_FACTORY).stringType().noDefaultValue();
- optionalOptions.add(keyTypeInfoFactoryOption);
-
- TypeInformation<?> keyTypeInfo =
- typeResolver.resolveKeyType(
- options, keyFormatOption, keyTypeInfoFactoryOption,
keyRowField);
-
- final Tuple2<Integer, List<StateValueColumnConfiguration>>
keyValueConfigProjections =
- Tuple2.of(
- keyValueProjections.f0,
- Arrays.stream(keyValueProjections.f1)
- .mapToObj(
- columnIndex ->
- createStateColumnConfiguration(
- columnIndex,
- rowType,
- options,
- optionalOptions,
- typeResolver))
- .collect(Collectors.toList()));
- FactoryUtil.validateFactoryOptions(requiredOptions, optionalOptions,
options);
+ // Validate schema and register per-field options eagerly (no class
loading) so that
+ // option validation passes at planning time.
+ int keyColumnIndex =
+ StateTableMapping.validateAndExtractKeyColumn(
+ context.getCatalogTable(), optionalOptions);
- Set<String> consumedOptionKeys = new HashSet<>();
- consumedOptionKeys.add(CONNECTOR.key());
-
requiredOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
-
optionalOptions.stream().map(ConfigOption::key).forEach(consumedOptionKeys::add);
- FactoryUtil.validateUnconsumedKeys(
- factoryIdentifier(), options.keySet(), consumedOptionKeys);
+ validateOptions(options, requiredOptions, optionalOptions);
+
+ // Defer I/O and class loading to scan time by creating the
StateTableMapping lazily.
+ Supplier<StateTableMapping> mappingSupplier =
+ () ->
+ StateTableMapping.from(
+ context.getCatalogTable(),
+ options,
+ statePath,
+ operatorIdentifier,
+ serializerConfig);
- return new SavepointDynamicTableSource(
+ RowType rowType = (RowType)
context.getPhysicalRowDataType().getLogicalType();
+
+ return new SavepointDynamicTableSource<>(
stateBackendType,
statePath,
operatorIdentifier,
- keyTypeInfo,
- keyValueConfigProjections,
- rowType);
+ keyColumnIndex,
+ mappingSupplier,
+ rowType,
+ "Savepoint Table Source",
+ SavepointDataStreamScanProvider::new);
}
- private StateValueColumnConfiguration createStateColumnConfiguration(
- int columnIndex,
- RowType rowType,
+ /**
+ * Creates a {@link FlattenedSavepointDynamicTableSource} for a table
exposing a single
+ * flattened LIST/MAP state (selected via {@link
SavepointConnectorOptions#STATE_READER_MODE}
+ * being set to {@link
SavepointConnectorOptions.StateReaderMode#KEYED_FLAT}). The state name is
+ * resolved from {@link SavepointConnectorOptions#FLATTENED_STATE_NAME}.
+ */
+ private DynamicTableSource createFlattenedDynamicTableSource(
+ Context context,
Configuration options,
- Set<ConfigOption<?>> optionalOptions,
- SavepointTypeInfoResolver typeResolver) {
-
- RowType.RowField valueRowField = rowType.getFields().get(columnIndex);
-
- ConfigOption<String> stateNameOption =
- optionOf(valueRowField.getName(),
STATE_NAME).stringType().noDefaultValue();
- optionalOptions.add(stateNameOption);
-
- ConfigOption<SavepointConnectorOptions.StateType> stateTypeOption =
- optionOf(valueRowField.getName(), STATE_TYPE)
- .enumType(SavepointConnectorOptions.StateType.class)
- .noDefaultValue();
- optionalOptions.add(stateTypeOption);
-
- ConfigOption<String> mapKeyFormatOption =
- optionOf(valueRowField.getName(),
KEY_CLASS).stringType().noDefaultValue();
- optionalOptions.add(mapKeyFormatOption);
-
- ConfigOption<String> mapKeyTypeInfoFactoryOption =
- optionOf(valueRowField.getName(),
KEY_TYPE_FACTORY).stringType().noDefaultValue();
- optionalOptions.add(mapKeyTypeInfoFactoryOption);
-
- ConfigOption<String> valueFormatOption =
- optionOf(valueRowField.getName(),
VALUE_CLASS).stringType().noDefaultValue();
- optionalOptions.add(valueFormatOption);
-
- ConfigOption<String> valueTypeInfoFactoryOption =
- optionOf(valueRowField.getName(),
VALUE_TYPE_FACTORY).stringType().noDefaultValue();
- optionalOptions.add(valueTypeInfoFactoryOption);
-
- LogicalType valueLogicalType = valueRowField.getType();
+ SerializerConfig serializerConfig,
+ String stateBackendType,
+ String statePath,
+ OperatorIdentifier operatorIdentifier) {
SavepointConnectorOptions.StateType stateType =
- options.getOptional(stateTypeOption)
- .orElseGet(() -> inferStateType(valueLogicalType));
+
FlattenedStateTableMapping.validateFlattenedSchema(context.getCatalogTable());
- TypeSerializer<?> mapKeyTypeSerializer =
- typeResolver.resolveSerializer(
- options,
- mapKeyFormatOption,
- mapKeyTypeInfoFactoryOption,
- valueRowField,
-
stateType.equals(SavepointConnectorOptions.StateType.MAP),
- SavepointTypeInfoResolver.InferenceContext.MAP_KEY);
-
- TypeSerializer<?> valueTypeSerializer =
- typeResolver.resolveSerializer(
- options,
- valueFormatOption,
- valueTypeInfoFactoryOption,
- valueRowField,
- true,
- SavepointTypeInfoResolver.InferenceContext.VALUE);
-
- return new StateValueColumnConfiguration(
- columnIndex,
-
options.getOptional(stateNameOption).orElse(valueRowField.getName()),
- stateType,
- mapKeyTypeSerializer,
- valueTypeSerializer);
- }
-
- private static ConfigOptions.OptionBuilder optionOf(String rowField,
String optionName) {
- return ConfigOptions.key(String.format("%s.%s.%s", FIELDS, rowField,
optionName));
- }
-
- private Tuple2<Integer, int[]>
createKeyValueProjections(ResolvedCatalogTable catalogTable) {
- ResolvedSchema schema = catalogTable.getResolvedSchema();
- if (schema.getPrimaryKey().isEmpty()) {
- throw new ValidationException("Could not find the primary key in
the table schema.");
- }
-
- List<String> keyFields = schema.getPrimaryKey().get().getColumns();
- if (keyFields.size() != 1) {
- throw new ValidationException(
- "Only a single primary key must be defined in the table
schema.");
- }
+ RowType rowType = (RowType)
context.getPhysicalRowDataType().getLogicalType();
- DataType physicalDataType = schema.toPhysicalRowDataType();
- int keyProjection = createKeyFormatProjection(physicalDataType,
keyFields.get(0));
- int[] valueProjection = createValueFormatProjection(physicalDataType,
keyProjection);
+ String stateName = validateAndGetFlattenedStateName(options);
- return Tuple2.of(keyProjection, valueProjection);
- }
-
- private int createKeyFormatProjection(DataType physicalDataType, String
keyField) {
- final LogicalType physicalType = physicalDataType.getLogicalType();
- Preconditions.checkArgument(
- physicalType.is(LogicalTypeRoot.ROW), "Row data type
expected.");
- final List<String> physicalFields =
LogicalTypeChecks.getFieldNames(physicalType);
- return physicalFields.indexOf(keyField);
- }
-
- private int[] createValueFormatProjection(DataType physicalDataType, int
keyProjection) {
- final LogicalType physicalType = physicalDataType.getLogicalType();
- Preconditions.checkArgument(
- physicalType.is(LogicalTypeRoot.ROW), "Row data type
expected.");
- final int physicalFieldCount =
LogicalTypeChecks.getFieldCount(physicalType);
- final IntStream physicalFields = IntStream.range(0,
physicalFieldCount);
+ // Defer I/O to scan time by creating the mapping lazily.
+ Supplier<FlattenedStateTableMapping> mappingSupplier =
+ () ->
+ FlattenedStateTableMapping.from(
+ context.getCatalogTable(),
+ stateName,
+ statePath,
+ operatorIdentifier,
+ serializerConfig,
+ stateType);
- return physicalFields.filter(pos -> keyProjection != pos).toArray();
+ return new FlattenedSavepointDynamicTableSource<>(
+ stateBackendType,
+ statePath,
+ operatorIdentifier,
+ FlattenedStateTableMapping.STATE_KEY_COLUMN_INDEX,
+ mappingSupplier,
+ rowType,
+ "Flattened Savepoint Table Source",
+ FlattenedSavepointDataStreamScanProvider::new);
}
- private SavepointConnectorOptions.StateType inferStateType(LogicalType
logicalType) {
- switch (logicalType.getTypeRoot()) {
- case ARRAY:
- return SavepointConnectorOptions.StateType.LIST;
+ /**
+ * Validates {@code options} against the required/optional option sets
extended with {@link
+ * SavepointConnectorOptions#FLATTENED_STATE_NAME}, and returns the
resolved state name — shared
+ * by every table kind whose columns represent a single named state's
flattened value fields (or
+ * a single scalar value column) rather than encoding the state's name via
the column layout
+ * itself.
+ */
+ private String validateAndGetFlattenedStateName(Configuration options) {
Review Comment:
FLATTENED_STATE_NAME is added ad-hoc inside
validateAndGetFlattenedStateName(), never through
requiredOptions()/optionalOptions() like every other option here. Anything
introspecting this factory's declared options directly (docs tooling, generic
Table API code) would never learn it exists or is required for KEYED_FLAT.
Worth registering it declaratively too?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input.deserializer;
+
+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.api.java.typeutils.runtime.PojoDeserializerCompatibilitySnapshot;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import
org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A {@link TypeSerializer} that reads the POJO binary format written by {@link
+ * org.apache.flink.api.java.typeutils.runtime.PojoSerializer} and produces
{@link GenericRowData}.
+ *
+ * <p>This deserializer does <em>not</em> require the user POJO class to be on
the classpath. It
+ * mirrors the exact binary protocol of {@code PojoSerializer}:
+ *
+ * <pre>{@code
+ * 1 byte: flags (bitmask)
+ * 0x01 IS_NULL → value is null, return null
+ * 0x02 NO_SUBCLASS → exact POJO class: read numFields × (isNull
boolean + field bytes)
+ * 0x08 IS_TAGGED_SUBCLASS → 1 byte subclass tag; delegate to registered
subclass deserializer
+ * 0x04 IS_SUBCLASS → UTF class name (must be read); Kryo not
supported → throws IOException
+ * }</pre>
+ *
+ * <p>Use {@link #create(PojoSerializerSnapshot)} to build an instance from a
savepoint snapshot.
+ */
+@Internal
+public final class PojoToRowDataDeserializer extends TypeSerializer<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PojoToRowDataDeserializer.class);
+
+ // Mirrors constants in PojoSerializer
+ static final int IS_NULL = 0x01;
+ static final int NO_SUBCLASS = 0x02;
+ static final int IS_SUBCLASS = 0x04;
+ static final int IS_TAGGED_SUBCLASS = 0x08;
Review Comment:
Can we share somehow this internal logic (`isNulll(...)`, or something like
that) instead of duplicating it?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/catalog/StateCatalog.java:
##########
@@ -0,0 +1,701 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
+import org.apache.flink.state.api.OperatorIdentifier;
+import org.apache.flink.state.api.StateTableUtils;
+import org.apache.flink.state.api.runtime.SavepointLoader;
+import org.apache.flink.state.api.schema.KeyedStateSchemaInfo;
+import org.apache.flink.state.table.SavepointConnectorOptions.StateReaderMode;
+import org.apache.flink.state.table.SavepointConnectorOptions.StateType;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogView;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import
org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+
+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.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * A read-only Flink SQL catalog that discovers checkpoints and savepoints
from a configured set of
+ * directories and exposes their metadata as queryable SQL databases and views.
+ *
+ * <p>The catalog maps Flink's three-level hierarchy as follows:
+ *
+ * <ul>
+ * <li>Catalog: the name given at registration time (e.g. {@code "state"})
+ * <li>Database: one entry per discovered snapshot (e.g. {@code
"app1_savepoint-acce1cedsad"})
+ * <li>Table: a single view named {@code "metadata"} per database, backed by
the {@code
+ * savepoint_metadata} function from {@code StateModule}
+ * </ul>
+ *
+ * <p>Database names preserve hyphens from the original directory names.
Backtick quoting is
+ * required in SQL for identifiers containing hyphens:
+ *
+ * <pre>{@code
+ * USE CATALOG state;
+ * USE `app1_savepoint-acce1cedsad`;
+ * SELECT * FROM metadata;
+ * }</pre>
+ *
+ * <p>{@code StateModule} must be loaded before querying any {@code metadata}
view:
+ *
+ * <pre>{@code
+ * tableEnv.loadModule("state", StateModule.INSTANCE);
+ * }</pre>
+ *
+ * <p>Each catalog operation fetches state on demand. {@link #listDatabases()}
performs a full
+ * directory scan; all other operations perform a single file check on the
specific snapshot path
+ * reconstructed from the database name. There is no background polling and no
shared cache.
+ *
+ * <p>All write operations throw {@link UnsupportedOperationException}.
+ */
+@PublicEvolving
+public class StateCatalog extends AbstractCatalog {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(StateCatalog.class);
+
+ public static final String METADATA_TABLE = "metadata";
+ public static final String OPERATOR_UID_PREFIX = "uid_";
+ public static final String OPERATOR_ID_PREFIX = "id_";
+ public static final String OPERATOR_TABLE_SUFFIX = "_keyed";
+ public static final String FLAT_STATE_TABLE_SUFFIX = "_keyed_flat";
+
+ private static final CatalogDatabase EMPTY_DATABASE =
+ new CatalogDatabaseImpl(Collections.emptyMap(), "");
+
+ private final SnapshotDiscovery discovery;
+
+ public StateCatalog(String name, Map<String, String> labelsToDirs) {
+ this(name, labelsToDirs,
StateCatalogOptions.LISTING_PARALLELISM.defaultValue());
+ }
+
+ public StateCatalog(String name, Map<String, String> labelsToDirs, int
listingParallelism) {
+ this(
+ name,
+ labelsToDirs,
+ listingParallelism,
+ StateCatalogOptions.DB_NAME_INCLUDE_TS.defaultValue());
+ }
+
+ public StateCatalog(
+ String name,
+ Map<String, String> labelsToDirs,
+ int listingParallelism,
+ boolean dbNameIncludeTs) {
+ super(name, "default");
+ this.discovery = new SnapshotDiscovery(labelsToDirs,
listingParallelism, dbNameIncludeTs);
+ }
+
+ @Override
+ @Nullable
+ public String getDefaultDatabase() {
+ return null;
+ }
+
+ //
-------------------------------------------------------------------------
+ // Lifecycle
+ //
-------------------------------------------------------------------------
+
+ @Override
+ public void open() throws CatalogException {
+ discovery.start();
+ listDatabases();
+ }
+
+ @Override
+ public void close() throws CatalogException {
+ discovery.stop();
+ }
+
+ //
-------------------------------------------------------------------------
+ // Databases
+ //
-------------------------------------------------------------------------
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ try {
+ return discovery.list();
+ } catch (IOException e) {
+ LOG.warn("Failed to list databases in catalog '{}'", getName(), e);
+ return Collections.emptyList();
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (discovery.find(databaseName).isEmpty()) {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ return EMPTY_DATABASE;
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) throws CatalogException
{
+ return discovery.find(databaseName).isPresent();
+ }
+
+ @Override
+ public void createDatabase(String name, CatalogDatabase database, boolean
ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ throw new UnsupportedOperationException("StateCatalog is read-only.");
+ }
+
+ @Override
+ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean
cascade)
+ throws DatabaseNotExistException, DatabaseNotEmptyException,
CatalogException {
+ throw new UnsupportedOperationException("StateCatalog is read-only.");
+ }
+
+ @Override
+ public void alterDatabase(String name, CatalogDatabase newDatabase,
boolean ignoreIfNotExists)
+ throws DatabaseNotExistException, CatalogException {
+ throw new UnsupportedOperationException("StateCatalog is read-only.");
+ }
+
+ //
-------------------------------------------------------------------------
+ // Tables and views
+ //
-------------------------------------------------------------------------
+
+ @Override
+ public List<String> listTables(String databaseName)
Review Comment:
listTables() never includes METADATA_TABLE, only real state-derived tables.
Catalog.listTables()'\''s own javadoc says it should return all tables, views
and materialized tables, views included. So SHOW VIEWS finds metadata but SHOW
TABLES will not, even though getTable() handles it fine directly. Why is it
excluded from listTables()?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input.deserializer;
+
+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.api.java.typeutils.runtime.PojoDeserializerCompatibilitySnapshot;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import
org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A {@link TypeSerializer} that reads the POJO binary format written by {@link
+ * org.apache.flink.api.java.typeutils.runtime.PojoSerializer} and produces
{@link GenericRowData}.
+ *
+ * <p>This deserializer does <em>not</em> require the user POJO class to be on
the classpath. It
+ * mirrors the exact binary protocol of {@code PojoSerializer}:
+ *
+ * <pre>{@code
+ * 1 byte: flags (bitmask)
+ * 0x01 IS_NULL → value is null, return null
+ * 0x02 NO_SUBCLASS → exact POJO class: read numFields × (isNull
boolean + field bytes)
+ * 0x08 IS_TAGGED_SUBCLASS → 1 byte subclass tag; delegate to registered
subclass deserializer
+ * 0x04 IS_SUBCLASS → UTF class name (must be read); Kryo not
supported → throws IOException
+ * }</pre>
+ *
+ * <p>Use {@link #create(PojoSerializerSnapshot)} to build an instance from a
savepoint snapshot.
+ */
+@Internal
+public final class PojoToRowDataDeserializer extends TypeSerializer<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PojoToRowDataDeserializer.class);
+
+ // Mirrors constants in PojoSerializer
+ static final int IS_NULL = 0x01;
+ static final int NO_SUBCLASS = 0x02;
+ static final int IS_SUBCLASS = 0x04;
+ static final int IS_TAGGED_SUBCLASS = 0x08;
+
+ private final int numFields;
+ private final TypeSerializer<?>[] fieldDeserializers;
+ private final LogicalType[] fieldTypes;
+ private final String[] fieldNames;
+ private final List<PojoToRowDataDeserializer>
registeredSubclassDeserializers;
+
+ /**
+ * Builds a {@link PojoToRowDataDeserializer} from a {@link
PojoSerializerSnapshot}.
+ *
+ * <p>For each field:
+ *
+ * <ul>
+ * <li>If the field snapshot is itself a {@link PojoSerializerSnapshot},
this method recurses
+ * to build a nested {@link PojoToRowDataDeserializer}.
+ * <li>For all other field types, the field's original serializer is
restored via {@link
+ * TypeSerializerSnapshot#restoreSerializer()}.
+ * </ul>
+ *
+ * <p>Registered subclasses are handled by building a deserializer for
each registered subclass
+ * snapshot in order (matching the tag index used in the binary format).
+ *
+ * @throws IllegalStateException if a required field serializer snapshot
is absent
+ */
+ public static PojoToRowDataDeserializer create(PojoSerializerSnapshot<?>
snapshot) {
+ List<AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>>
fieldEntries =
+ snapshot.getFieldSnapshotEntries();
+
+ List<TypeSerializer<?>> fieldDeserializerList = new
ArrayList<>(fieldEntries.size());
+ List<LogicalType> fieldTypeList = new ArrayList<>(fieldEntries.size());
+ List<String> fieldNameList = new ArrayList<>(fieldEntries.size());
+
+ for (AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>> entry
: fieldEntries) {
+ String fieldName = entry.getKey();
+ TypeSerializerSnapshot<?> fieldSnapshot = entry.getValue();
+
+ if (fieldSnapshot == null) {
Review Comment:
EnumSerializerSnapshot was not patched with CustomRestoreSerializerFactory
support, its readSnapshot() still throws eagerly on a missing enum class.
Unlike Kryo, this does not seem to be a stated limitation, and it fails at
table-creation time (fieldSnapshot == null here), not per-row. Was enum
considered?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java:
##########
@@ -91,6 +91,21 @@ <N, S extends State, T> void applyToAllKeys(
*/
<N> Stream<K> getKeys(List<String> states, N namespace);
+ /**
+ * @return A stream of all keys for the multiple states and a given
namespace, paired with the
Review Comment:
This javadoc requires knowing about classless deserialization to parse,
shouldn't be necessary for a core, widely-implemented interface. I can be more
self-contained: `returns the key-group each key is actually stored under,
instead of recomputed via hashCode. Use when the key object's hash isn't
guaranteed to match what wrote the data.` The classless-deserialization detail
can be moved to the call site in `MultiStateKeyIterator` instead.
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java:
##########
@@ -229,19 +226,37 @@ public void open(KeyGroupRangeInputSplit split) throws
IOException {
stateBackend,
executionConfig)
.withMaxParallelism(split.getNumKeyGroups())
- .withKey(operator,
runtimeContext.createSerializer(operator.getKeyType()))
- .build(LOG);
+ .withKey(operator,
runtimeContext.createSerializer(operator.getKeyType()));
+
+ // Deserialize any POJO/Avro state whose class is missing from the
classpath into
+ // RowData/GenericRecord instead of failing the restore.
PojoSerializerSnapshot and
+ // AvroSerializerSnapshot only consult this factory once they've
already determined that the
+ // class they need is genuinely missing, so registering it
unconditionally is safe and has
+ // no
+ // effect on states whose classes are present.
+ //
+ // The factory is never cleared: it is registered on the task thread
that also drives all
+ // subsequent state access, and a given type is either on the
classpath for the whole read
+ // or
+ // not at all. Clearing it after build() would break the RocksDB
backend, which defers
+ // schema
+ // compatibility checks to the first state access in nextRecord().
+
CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create);
Review Comment:
remove()'s own javadoc says to clear it after restore, to avoid leaking into
a reused thread. Somehow there is an inconsistency then, right? `The factory is
never cleared`
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java:
##########
@@ -0,0 +1,641 @@
+/*
+ * 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.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.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.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.warn(
+ "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.warn(
+ "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);
+ options.put(
+ SavepointConnectorOptions.STATE_READER_MODE.key(),
+ (windowType == null
+ ?
SavepointConnectorOptions.StateReaderMode.KEYED
+ :
SavepointConnectorOptions.StateReaderMode.WINDOWED)
+ .toString());
+ withStateBackendType(options, metadata, operatorIdentifier);
+
+ return
CatalogTable.newBuilder().schema(schema).options(options).build();
+ }
+
+ /**
+ * Builds a {@link CatalogTable} exposing a single keyed LIST or MAP state
flattened into one
+ * row per list element / map entry, rather than one row per key.
+ *
+ * <p>The resulting table has 3 columns, with a composite primary key on
{@code state_key} and
+ * the sub-key column (the {@code state_key} value repeats across rows
belonging to the same
+ * key, but the pair uniquely identifies a row). The third column has a
fixed name — not the
+ * state's own name, to avoid collisions with other (reserved) column
names:
+ *
+ * <ul>
+ * <li>LIST: {@code (state_key, list_index, list_value)}, primary key
{@code (state_key,
+ * list_index)}
+ * <li>MAP: {@code (state_key, map_key, map_value)}, primary key {@code
(state_key, map_key)}
+ * </ul>
+ *
+ * @param metadata the checkpoint metadata the operator belongs to
+ * @param schemaInfo the schema information returned by {@link
#getKeyedStateSchema}
+ * @param stateName the name of the LIST or MAP state to flatten
+ * @param statePath the path to the savepoint / checkpoint
+ * @param operatorIdentifier identifies the operator whose state to read
+ * @return a {@link CatalogTable} ready for registration
+ */
+ public static CatalogTable getFlattenedStateCatalogTable(
+ CheckpointMetadata metadata,
+ KeyedStateSchemaInfo schemaInfo,
+ String stateName,
+ String statePath,
+ OperatorIdentifier operatorIdentifier) {
+ return buildFlattenedKeyedCatalogTable(
+ metadata, schemaInfo, stateName, statePath,
operatorIdentifier, false);
+ }
+
+ /**
+ * Builds a {@link CatalogTable} exposing a single LIST or MAP state
flattened into one row per
+ * list element / map entry, either plain-keyed ({@code windowed ==
false}, see {@link
+ * #getFlattenedStateCatalogTable}) or namespaced ({@code windowed ==
true}).
+ */
+ private static CatalogTable buildFlattenedKeyedCatalogTable(
+ CheckpointMetadata metadata,
+ KeyedStateSchemaInfo schemaInfo,
+ String stateName,
+ String statePath,
+ OperatorIdentifier operatorIdentifier,
+ boolean windowed) {
+
+ KeyedStateSchemaInfo.StateEntryInfo entryInfo =
schemaInfo.stateSchemas.get(stateName);
+ if (entryInfo == null) {
+ throw new IllegalArgumentException(
+ "State '"
+ + stateName
+ + "' not found for operator '"
+ + operatorIdentifier
+ + "'.");
+ }
+ if (entryInfo.stateType != SavepointConnectorOptions.StateType.LIST
+ && entryInfo.stateType !=
SavepointConnectorOptions.StateType.MAP) {
+ throw new IllegalArgumentException(
+ "Flattened state tables are only supported for LIST and
MAP states, but '"
+ + stateName
+ + "' is "
+ + entryInfo.stateType
+ + ".");
+ }
+ if (windowed && entryInfo.windowLogicalType == null) {
+ throw new IllegalArgumentException(
+ "State '"
+ + stateName
+ + "' is not a namespaced state for operator '"
+ + operatorIdentifier
+ + "'.");
+ }
+
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column(
+ "state_key",
LogicalTypeDataTypeConverter.toDataType(schemaInfo.keyType).notNull());
+ if (windowed) {
+ schemaBuilder.column(
+ "state_window",
+
LogicalTypeDataTypeConverter.toDataType(entryInfo.windowLogicalType).notNull());
+ }
+
+ String subKeyColumnName = addFlattenedValueColumns(schemaBuilder,
entryInfo);
+ if (!windowed) {
+ schemaBuilder.primaryKeyNamed(
+ "PK_state_key_" + subKeyColumnName, "state_key",
subKeyColumnName);
+ }
+ Schema schema = schemaBuilder.build();
+
+ Map<String, String> options = buildBaseConnectorOptions(statePath,
operatorIdentifier);
+ options.put(
+ SavepointConnectorOptions.STATE_READER_MODE.key(),
+ (windowed
+ ?
SavepointConnectorOptions.StateReaderMode.WINDOWED_FLAT
+ :
SavepointConnectorOptions.StateReaderMode.KEYED_FLAT)
+ .toString());
+ options.put(SavepointConnectorOptions.FLATTENED_STATE_NAME.key(),
stateName);
+ withStateBackendType(options, metadata, operatorIdentifier);
+
+ return
CatalogTable.newBuilder().schema(schema).options(options).build();
+ }
+
+ /**
+ * Adds the LIST- or MAP-shaped sub-key and value columns (e.g. {@code
(list_index, list_value)}
+ * or {@code (map_key, map_value)}) for a flattened state table, and
returns the sub-key
+ * column's name.
+ */
+ private static String addFlattenedValueColumns(
+ Schema.Builder schemaBuilder, KeyedStateSchemaInfo.StateEntryInfo
entryInfo) {
+ LogicalType valueType;
+ String subKeyColumnName;
+ String valueColumnName;
+ if (entryInfo.stateType == SavepointConnectorOptions.StateType.LIST) {
+ valueType = ((ArrayType) entryInfo.logicalType).getElementType();
+ subKeyColumnName = "list_index";
+ valueColumnName = "list_value";
+ schemaBuilder.column(
+ subKeyColumnName,
+ LogicalTypeDataTypeConverter.toDataType(new
BigIntType(false)));
+ } else {
+ MapType mapType = (MapType) entryInfo.logicalType;
+ valueType = mapType.getValueType();
+ subKeyColumnName = "map_key";
+ valueColumnName = "map_value";
+ schemaBuilder.column(
+ subKeyColumnName,
+
LogicalTypeDataTypeConverter.toDataType(mapType.getKeyType()).notNull());
+ }
+ schemaBuilder.column(valueColumnName,
LogicalTypeDataTypeConverter.toDataType(valueType));
+ return subKeyColumnName;
+ }
+
+ //
-------------------------------------------------------------------------
+ // 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) {
Review Comment:
Both literals are unlinked duplicates: _timer_state mirrors a
package-private constant in InternalTimeServiceManagerImpl, merging-window-set
is not even named anywhere, it is inline in WindowOperator. Either changing
there silently breaks this, with nothing to catch it. Worth hardening this
coupling somehow?
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java:
##########
@@ -206,7 +212,17 @@ void writeSnapshotData(DataOutputView out) throws
IOException {
private static <T> PojoSerializerSnapshotData<T> readSnapshotData(
DataInputView in, ClassLoader userCodeClassLoader) throws
IOException {
- Class<T> pojoClass = InstantiationUtil.resolveClassByName(in,
userCodeClassLoader);
+ final String pojoClassName = in.readUTF();
+ Class<T> pojoClass = null;
Review Comment:
TLDR: If condition can be added to both places
1. `PojoSerializerSnapshotData.readSnapshotData()` (ClassNotFoundException
e) block:
```
if (CustomRestoreSerializerFactory.get() == null) {
throw new IOException("Could not find class '" + pojoClassName + "'
in classpath.", e);
}
```
2. `AvroSerializerSnapshot.tryFindClass()` (ClassNotFoundException e) block:
```
if (CustomRestoreSerializerFactory.get() == null) {
throw new IllegalStateException("Unable to find the class '" +
className + "' ... Was the class moved or renamed?", e);
}
```
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java:
##########
@@ -0,0 +1,641 @@
+/*
+ * 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.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.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.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.warn(
+ "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) {
Review Comment:
convert()'s contract documents UnsupportedOperationException for unsupported
types, catching that specifically would still handle the legitimate case. Why
catch (Exception e) here? It also absorbs a genuine bug (NPE,
ClassCastException) into the same silent column missing outcome.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
