diff --git a/docs/content/docs/libs/state_processor_api.md b/docs/content/docs/libs/state_processor_api.md
index 0b0472b7f39e40..336502ab3f2173 100644
--- a/docs/content/docs/libs/state_processor_api.md
+++ b/docs/content/docs/libs/state_processor_api.md
@@ -749,17 +749,13 @@ The following predicates on the key column can be pushed down:
| Option | Required | Default | Type | Description |
|----------------------------------|----------|---------|----------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| fields.#.state-name | optional | (none) | String | Overrides the state name which must be used for state reading. This can be useful when the state name contains characters which are not compliant with SQL column names. |
-| fields.#.state-type | optional | (none) | Enum Possible values: list, map, value | Defines the state type which must be used for state reading, including value, list and map. When it's not provided then it tries to infer from the SQL type (ARRAY=list, MAP=map, all others=value). |
-| fields.#.key-class | optional | (none) | String | Defines the format class scheme for decoding map key data (for ex. java.lang.Long). Either key-class or key-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported). |
-| fields.#.key-type-factory | optional | (none) | String | Defines the type information factory for decoding map key data. Either key-class or key-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported). |
-| fields.#.value-class | optional | (none) | String | Defines the format class scheme for decoding value data (for ex. java.lang.Long). Either value-class or value-info-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported). |
-| fields.#.value-type-factory | optional | (none) | String | Defines the type information factory for decoding value data. Either value-class or value-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported). |
### Default Data Type Mapping
-The state SQL connector infers the data type for primitive types when `fields.#.value-class` and `fields.#.key-class`
-are not defined. The following table shows the `Flink SQL type` -> `Java type` default mapping. If the mapping is not calculated properly
-then it can be overridden with the two mentioned config parameters on a per-column basis.
+The state SQL connector automatically infers each column's data type from the savepoint's serializer
+snapshot metadata (this covers primitive, Avro, Row and POJO types). When no serializer snapshot is
+available for a column, it falls back to inferring a primitive Java type directly from the SQL type,
+using the mapping below.
| Flink SQL type | Java type |
|-------------------------|-------------------------------------------------------------------------|
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CustomRestoreSerializerFactory.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CustomRestoreSerializerFactory.java
new file mode 100644
index 00000000000000..3ad8c0870f0044
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CustomRestoreSerializerFactory.java
@@ -0,0 +1,118 @@
+/*
+ * 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.common.typeutils;
+
+import org.apache.flink.annotation.Internal;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.function.Function;
+
+/**
+ * Thread-scoped holder for a factory that builds a fallback {@link TypeSerializer} when a {@link
+ * TypeSerializerSnapshot} restores itself but a class it depends on is not on the classpath — for
+ * example a POJO's declared type, or an Avro record's specific/reflect runtime type.
+ *
+ *
This exists solely to support the Flink State Processing API's ability to read state whose
+ * original classes are not on the classpath (e.g. converting savepoint state into table rows
+ * without the user's job JAR). It must never be set by, or otherwise affect, regular job restores:
+ * a {@code TypeSerializerSnapshot} only ever consults this factory after it has already determined
+ * — independently of this class — that the class it needs is genuinely missing, and only when a
+ * factory has actually been registered. With no factory registered (the default for every job that
+ * is not using the State Processing API), behavior is unchanged from before this hook existed: the
+ * snapshot fails fast with a {@code ClassNotFoundException} or equivalent.
+ *
+ *
A {@code ThreadLocal} is used because {@link
+ * CompositeTypeSerializerSnapshot#restoreSerializer()} eagerly restores all of its nested
+ * serializers and offers no hook for substituting one of them, so a POJO or Avro type nested
+ * arbitrarily deep inside a composite snapshot (e.g. a list or map serializer snapshot) cannot be
+ * reached otherwise.
+ */
+@Internal
+public final class CustomRestoreSerializerFactory {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CustomRestoreSerializerFactory.class);
+
+ private static final ThreadLocal, TypeSerializer>>>
+ FACTORY = new ThreadLocal<>();
+
+ private CustomRestoreSerializerFactory() {}
+
+ /** Registers the fallback factory for the current thread. */
+ public static void set(Function, TypeSerializer>> factory) {
+ FACTORY.set(factory);
+ }
+
+ /** Returns the fallback factory registered via {@link #set}, or {@code null} if none. */
+ public static Function, TypeSerializer>> get() {
+ return FACTORY.get();
+ }
+
+ /** Clears the fallback factory registered for the current thread. */
+ public static void remove() {
+ FACTORY.remove();
+ }
+
+ /**
+ * Resolves {@code className} via {@code classLoader}, or returns {@code null} if it cannot be
+ * found and a fallback factory is registered for the current thread.
+ *
+ * @throws NoClassDefFoundError if the class cannot be found and no fallback factory is
+ * registered.
+ */
+ @SuppressWarnings("unchecked")
+ public static Class resolveOrNull(String className, ClassLoader classLoader) {
+ try {
+ return (Class) Class.forName(className, false, classLoader);
+ } catch (ClassNotFoundException e) {
+ if (get() == null) {
+ throw missingClass(className, e);
+ }
+ LOG.debug(
+ "Class '{}' not found on classpath; a CustomRestoreSerializerFactory is"
+ + " registered to read the data without it.",
+ className);
+ return null;
+ }
+ }
+
+ /**
+ * Builds the fallback serializer for a {@code snapshot} whose runtime class, {@code
+ * missingClassName}, could not be loaded, using the factory registered via {@link #set}.
+ *
+ * @throws NoClassDefFoundError if no factory is registered for the current thread.
+ */
+ @SuppressWarnings("unchecked")
+ public static TypeSerializer restoreFallbackSerializer(
+ TypeSerializerSnapshot snapshot, String missingClassName) {
+ Function, TypeSerializer>> fallback = get();
+ if (fallback == null) {
+ throw missingClass(missingClassName, new ClassNotFoundException(missingClassName));
+ }
+ return (TypeSerializer) fallback.apply(snapshot);
+ }
+
+ private static NoClassDefFoundError missingClass(
+ String className, ClassNotFoundException cause) {
+ NoClassDefFoundError error = new NoClassDefFoundError(className);
+ error.initCause(cause);
+ return error;
+ }
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
index ca7d1fc56326a9..ee5b5898ab2ca9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java
@@ -20,12 +20,12 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
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 org.apache.flink.util.InstantiationUtil;
import java.io.IOException;
import java.io.ObjectInputStream;
@@ -184,6 +184,8 @@ public static final class EnumSerializerSnapshot>
private T[] enums;
private Class enumClass;
+ private String enumClassName;
+ private String[] enumNames;
@SuppressWarnings("unused")
public EnumSerializerSnapshot() {
@@ -213,16 +215,23 @@ public void writeSnapshot(DataOutputView out) throws IOException {
@Override
public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
throws IOException {
- enumClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+ final String className = in.readUTF();
+ enumClass =
+ CustomRestoreSerializerFactory.resolveOrNull(className, userCodeClassLoader);
+ enumClassName = className;
int numEnumConstants = in.readInt();
-
@SuppressWarnings("unchecked")
- T[] previousEnums = (T[]) Array.newInstance(enumClass, numEnumConstants);
+ T[] previousEnums =
+ enumClass != null ? (T[]) Array.newInstance(enumClass, numEnumConstants) : null;
+ String[] names = new String[numEnumConstants];
for (int i = 0; i < numEnumConstants; i++) {
- String enumName = in.readUTF();
+ names[i] = in.readUTF();
+ if (previousEnums == null) {
+ continue;
+ }
try {
- previousEnums[i] = Enum.valueOf(enumClass, enumName);
+ previousEnums[i] = Enum.valueOf(enumClass, names[i]);
} catch (IllegalArgumentException e) {
throw new IllegalStateException(
"Could not create a restore serializer for enum "
@@ -230,14 +239,25 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
+ ". Probably because an enum value was removed.");
}
}
+ enumNames = names;
+ if (enumClass == null) {
+ return;
+ }
this.enums = previousEnums;
}
+ /** Returns the enum constant names in this snapshot, ordered by their wire ordinal. */
+ public String[] getEnumNames() {
+ return enumNames;
+ }
+
@Override
public TypeSerializer restoreSerializer() {
- checkState(enumClass != null, "Enum class can not be null.");
-
+ if (enumClass == null) {
+ return CustomRestoreSerializerFactory.restoreFallbackSerializer(
+ this, enumClassName);
+ }
return new EnumSerializer<>(enumClass, enums);
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
index 9d7e4254e861eb..19e31c9f211d3c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -48,10 +48,10 @@
public final class PojoSerializer extends TypeSerializer {
// Flags for the header
- private static final byte IS_NULL = 1;
- private static final byte NO_SUBCLASS = 2;
- private static final byte IS_SUBCLASS = 4;
- private static final byte IS_TAGGED_SUBCLASS = 8;
+ public static final byte IS_NULL = 1;
+ public static final byte NO_SUBCLASS = 2;
+ public static final byte IS_SUBCLASS = 4;
+ public static final byte IS_TAGGED_SUBCLASS = 8;
private static final long serialVersionUID = 1L;
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
index cb8deb5b9e7cc8..cdfe1b9b12ebae 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java
@@ -23,6 +23,7 @@
import org.apache.flink.api.common.serialization.SerializerConfigImpl;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil.IntermediateCompatibilityResult;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
@@ -34,10 +35,12 @@
import java.io.IOException;
import java.lang.reflect.Field;
+import java.util.AbstractMap.SimpleEntry;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -143,6 +146,11 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
@Override
@SuppressWarnings("unchecked")
public TypeSerializer restoreSerializer() {
+ if (snapshotData.getPojoClass() == null) {
+ return CustomRestoreSerializerFactory.restoreFallbackSerializer(
+ this, snapshotData.getPojoClassName());
+ }
+
final int numFields = snapshotData.getFieldSerializerSnapshots().size();
final ArrayList restoredFields = new ArrayList<>(numFields);
@@ -257,6 +265,56 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility(
return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
+ // ---------------------------------------------------------------------------------------------
+ // Schema extraction support
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Returns {@code true} if the POJO class could be loaded from the classloader that read this
+ * snapshot. When {@code false}, {@link #restoreSerializer()} delegates to the serializer
+ * supplied via {@link CustomRestoreSerializerFactory} instead of building a {@link
+ * PojoSerializer}.
+ */
+ @Internal
+ public boolean isPojoClassAvailable() {
+ return snapshotData.getPojoClass() != null;
+ }
+
+ /**
+ * Returns the POJO class name as stored in the snapshot. Available even when the class cannot
+ * be loaded.
+ */
+ @Internal
+ public String getPojoClassName() {
+ return snapshotData.getPojoClassName();
+ }
+
+ /**
+ * Returns an ordered list of (field name, field serializer snapshot) pairs. Field names are
+ * always present; snapshot values may be {@code null} when the field snapshot could not be
+ * read.
+ */
+ @Internal
+ public List>> getFieldSnapshotEntries() {
+ List>> result = new ArrayList<>();
+ snapshotData
+ .getFieldSerializerSnapshots()
+ .forEach(
+ (fieldName, field, fieldSnapshot) ->
+ result.add(new SimpleEntry<>(fieldName, fieldSnapshot)));
+ return result;
+ }
+
+ /**
+ * Returns the registered subclass serializer snapshots in tag order (tag 0, 1, 2, …). Values
+ * may be {@code null} if a subclass snapshot was not readable.
+ */
+ @Internal
+ public List> getRegisteredSubclassSnapshotsOrdered() {
+ return new ArrayList<>(
+ snapshotData.getRegisteredSubclassSerializerSnapshots().unwrapOptionals().values());
+ }
+
// ---------------------------------------------------------------------------------------------
// Utility methods
// ---------------------------------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
index 6b2bd112ad14f7..ceaaa3158e4693 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotData.java
@@ -19,12 +19,12 @@
package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.util.CollectionUtil;
-import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.LinkedOptionalMap;
import org.apache.flink.util.function.BiConsumerWithException;
import org.apache.flink.util.function.BiFunctionWithException;
@@ -32,6 +32,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.LinkedHashMap;
@@ -113,6 +115,7 @@ static PojoSerializerSnapshotData createFrom(
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClass.getName(),
fieldSerializerSnapshots,
optionalMapOf(registeredSubclassSerializerSnapshots, Class::getName),
optionalMapOf(nonRegisteredSubclassSerializerSnapshots, Class::getName));
@@ -153,12 +156,14 @@ static PojoSerializerSnapshotData createFrom(
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClass.getName(),
fieldSerializerSnapshots,
optionalMapOf(existingRegisteredSubclassSerializerSnapshots, Class::getName),
optionalMapOf(existingNonRegisteredSubclassSerializerSnapshots, Class::getName));
}
- private Class pojoClass;
+ @Nullable private Class pojoClass;
+ private String pojoClassName;
private LinkedOptionalMap> fieldSerializerSnapshots;
private LinkedOptionalMap, TypeSerializerSnapshot>>
registeredSubclassSerializerSnapshots;
@@ -166,14 +171,16 @@ static PojoSerializerSnapshotData createFrom(
nonRegisteredSubclassSerializerSnapshots;
private PojoSerializerSnapshotData(
- Class typeClass,
+ @Nullable Class typeClass,
+ String pojoClassName,
LinkedOptionalMap> fieldSerializerSnapshots,
LinkedOptionalMap, TypeSerializerSnapshot>>
registeredSubclassSerializerSnapshots,
LinkedOptionalMap, TypeSerializerSnapshot>>
nonRegisteredSubclassSerializerSnapshots) {
- this.pojoClass = checkNotNull(typeClass);
+ this.pojoClass = typeClass;
+ this.pojoClassName = checkNotNull(pojoClassName);
this.fieldSerializerSnapshots = checkNotNull(fieldSerializerSnapshots);
this.registeredSubclassSerializerSnapshots =
checkNotNull(registeredSubclassSerializerSnapshots);
@@ -186,7 +193,7 @@ private PojoSerializerSnapshotData(
// ---------------------------------------------------------------------------------------------
void writeSnapshotData(DataOutputView out) throws IOException {
- out.writeUTF(pojoClass.getName());
+ out.writeUTF(pojoClassName);
writeOptionalMap(
out,
fieldSerializerSnapshots,
@@ -206,7 +213,14 @@ void writeSnapshotData(DataOutputView out) throws IOException {
private static PojoSerializerSnapshotData readSnapshotData(
DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
- Class pojoClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+ final String pojoClassName = in.readUTF();
+ Class pojoClass =
+ CustomRestoreSerializerFactory.resolveOrNull(pojoClassName, userCodeClassLoader);
+ if (pojoClass == null) {
+ LOG.debug(
+ "POJO class '{}' not found on classpath; schema can still be read from field snapshots.",
+ pojoClassName);
+ }
LinkedOptionalMap> fieldSerializerSnapshots =
readOptionalMap(
@@ -226,6 +240,7 @@ private static PojoSerializerSnapshotData readSnapshotData(
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClassName,
fieldSerializerSnapshots,
registeredSubclassSerializerSnapshots,
nonRegisteredSubclassSerializerSnapshots);
@@ -235,10 +250,15 @@ private static PojoSerializerSnapshotData readSnapshotData(
// Snapshot data accessors
// ---------------------------------------------------------------------------------------------
+ @Nullable
Class getPojoClass() {
return pojoClass;
}
+ String getPojoClassName() {
+ return pojoClassName;
+ }
+
LinkedOptionalMap> getFieldSerializerSnapshots() {
return fieldSerializerSnapshots;
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMissingClassTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMissingClassTest.java
new file mode 100644
index 00000000000000..a485c7cf85b354
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/EnumSerializerSnapshotMissingClassTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.common.typeutils.base;
+
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.EnumSerializer.EnumSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for the lenient enum class-loading in {@link EnumSerializerSnapshot}: with a {@link
+ * CustomRestoreSerializerFactory} registered, an {@link EnumSerializerSnapshot} must be readable
+ * even when the enum class is not on the classpath, with the constant names (in wire-ordinal order)
+ * remaining accessible.
+ */
+class EnumSerializerSnapshotMissingClassTest {
+
+ enum SomeEnum {
+ FOO,
+ BAR,
+ BAZ
+ }
+
+ @Test
+ void testReadWithClassAbsent() throws IOException {
+ EnumSerializerSnapshot read;
+ CustomRestoreSerializerFactory.set(
+ snapshot -> {
+ throw new UnsupportedOperationException("not exercised in this test");
+ });
+ try {
+ read = roundtripSnapshot(writeSnapshot(), withoutEnumClassLoader());
+ } finally {
+ CustomRestoreSerializerFactory.remove();
+ }
+
+ assertThat(read.getEnumNames()).containsExactly("FOO", "BAR", "BAZ");
+ }
+
+ /**
+ * Regular job restores (i.e. without a {@link CustomRestoreSerializerFactory} registered, as is
+ * always the case outside of the State Processing API) must still fail fast when the enum class
+ * is genuinely missing.
+ */
+ @Test
+ void testReadFailsFastWithoutFallbackFactory() {
+ assertThatThrownBy(() -> roundtripSnapshot(writeSnapshot(), withoutEnumClassLoader()))
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasCauseInstanceOf(ClassNotFoundException.class);
+ }
+
+ /** Hides the enum class from the classloader used to read the snapshot back. */
+ private ClassLoader withoutEnumClassLoader() {
+ return new ClassLoader(getClass().getClassLoader()) {
+ @Override
+ protected Class> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.contains(SomeEnum.class.getSimpleName())) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+ }
+
+ private static EnumSerializerSnapshot writeSnapshot() {
+ return new EnumSerializer<>(SomeEnum.class).snapshotConfiguration();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static EnumSerializerSnapshot roundtripSnapshot(
+ EnumSerializerSnapshot snapshot, ClassLoader classLoader) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return (EnumSerializerSnapshot)
+ TypeSerializerSnapshot.readVersionedSnapshot(in, classLoader);
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotLenientReadTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotLenientReadTest.java
new file mode 100644
index 00000000000000..6f5fedb5bae511
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotLenientReadTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+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.api.java.typeutils.TypeExtractor;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for the lenient POJO class-loading in {@link PojoSerializerSnapshotData}: with a {@link
+ * CustomRestoreSerializerFactory} registered, a {@link PojoSerializerSnapshot} must be readable
+ * even when the POJO class is not on the classpath, with the class name, the field names, and the
+ * field serializer snapshots all remaining accessible.
+ */
+class PojoSerializerSnapshotLenientReadTest {
+
+ private static final Map> EXPECTED_FIELD_SNAPSHOTS =
+ Map.of(
+ "name", StringSerializer.StringSerializerSnapshot.class,
+ "age", IntSerializer.IntSerializerSnapshot.class,
+ "score", LongSerializer.LongSerializerSnapshot.class);
+
+ /** POJO present while the snapshot is written, hidden from the classloader while it is read. */
+ public static class SomePojo {
+ public String name;
+ public int age;
+ public long score;
+ }
+
+ @Test
+ void testReadWithClassPresent() throws IOException {
+ PojoSerializerSnapshot read =
+ roundtripSnapshot(writeSnapshot(), getClass().getClassLoader());
+
+ assertThat(read.isPojoClassAvailable()).isTrue();
+ assertThat(read.getPojoClassName()).isEqualTo(SomePojo.class.getName());
+ assertFieldSnapshots(read);
+ }
+
+ @Test
+ void testReadWithClassAbsent() throws IOException {
+ PojoSerializerSnapshot read;
+ CustomRestoreSerializerFactory.set(
+ snapshot -> {
+ throw new UnsupportedOperationException("not exercised in this test");
+ });
+ try {
+ read = roundtripSnapshot(writeSnapshot(), withoutPojoClassLoader());
+ } finally {
+ CustomRestoreSerializerFactory.remove();
+ }
+
+ assertThat(read.isPojoClassAvailable()).isFalse();
+ assertThat(read.getPojoClassName()).isEqualTo(SomePojo.class.getName());
+ // Field names stay available because the key name is written before the framed value.
+ assertFieldSnapshots(read);
+ }
+
+ /**
+ * Regular job restores (i.e. without a {@link CustomRestoreSerializerFactory} registered, as is
+ * always the case outside of the State Processing API) must still fail fast when the POJO class
+ * is genuinely missing, exactly as before lenient reading was introduced.
+ */
+ @Test
+ void testReadFailsFastWithoutFallbackFactory() {
+ assertThatThrownBy(() -> roundtripSnapshot(writeSnapshot(), withoutPojoClassLoader()))
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasCauseInstanceOf(ClassNotFoundException.class);
+ }
+
+ /**
+ * A {@link CustomRestoreSerializerFactory} only needs to be registered for the duration it is
+ * actually relied on: reading here succeeds with one present, but restoring a working
+ * serializer must still fail once it has been removed again.
+ */
+ @Test
+ void testRestoreSerializerFailsWithoutFallbackFactory() throws IOException {
+ PojoSerializerSnapshot read;
+ CustomRestoreSerializerFactory.set(
+ snapshot -> {
+ throw new UnsupportedOperationException("not exercised in this test");
+ });
+ try {
+ read = roundtripSnapshot(writeSnapshot(), withoutPojoClassLoader());
+ } finally {
+ CustomRestoreSerializerFactory.remove();
+ }
+
+ assertThat(read.isPojoClassAvailable()).isFalse();
+ assertThatThrownBy(read::restoreSerializer)
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasCauseInstanceOf(ClassNotFoundException.class);
+ }
+
+ /** Hides both the POJO class itself and the declaring class of each of its fields. */
+ private ClassLoader withoutPojoClassLoader() {
+ return new ClassLoader(getClass().getClassLoader()) {
+ @Override
+ protected Class> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.contains(SomePojo.class.getSimpleName())) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+ }
+
+ private static void assertFieldSnapshots(PojoSerializerSnapshot snapshot) {
+ List>> entries =
+ snapshot.getFieldSnapshotEntries();
+
+ assertThat(entries).hasSize(EXPECTED_FIELD_SNAPSHOTS.size());
+ for (SimpleEntry> entry : entries) {
+ Class> expectedType = EXPECTED_FIELD_SNAPSHOTS.get(entry.getKey());
+ assertThat(expectedType).as("unexpected field '%s'", entry.getKey()).isNotNull();
+ assertThat(entry.getValue())
+ .as("snapshot of field '%s'", entry.getKey())
+ .isExactlyInstanceOf(expectedType);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PojoSerializerSnapshot writeSnapshot() {
+ return (PojoSerializerSnapshot)
+ TypeExtractor.createTypeInfo(SomePojo.class)
+ .createSerializer(new SerializerConfigImpl())
+ .snapshotConfiguration();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PojoSerializerSnapshot roundtripSnapshot(
+ PojoSerializerSnapshot snapshot, ClassLoader classLoader) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return (PojoSerializerSnapshot)
+ TypeSerializerSnapshot.readVersionedSnapshot(in, classLoader);
+ }
+}
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
index 92580e71b2906d..4a097ed7e5d06e 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshot.java
@@ -19,6 +19,7 @@
package org.apache.flink.formats.avro.typeutils;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
@@ -33,7 +34,7 @@
import org.apache.avro.specific.SpecificData;
import org.apache.avro.specific.SpecificRecord;
-import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Objects;
@@ -49,7 +50,8 @@
* @param The data type that the originating serializer of this configuration serializes.
*/
public class AvroSerializerSnapshot implements TypeSerializerSnapshot {
- private Class runtimeType;
+ @Nullable private Class runtimeType;
+ private String runtimeTypeName;
private Schema schema;
private Schema runtimeSchema;
@@ -61,6 +63,7 @@ public AvroSerializerSnapshot() {
AvroSerializerSnapshot(Schema schema, Class runtimeType) {
this.schema = schema;
this.runtimeType = runtimeType;
+ this.runtimeTypeName = runtimeType.getName();
}
@Override
@@ -106,7 +109,11 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
private void readV1(DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
final String previousSchemaDefinition = in.readUTF();
this.schema = parseAvroSchema(previousSchemaDefinition);
- this.runtimeType = findClassOrFallbackToGeneric(userCodeClassLoader, schema.getFullName());
+ this.runtimeTypeName = schema.getFullName();
+ // V1 snapshots predate CustomRestoreSerializerFactory support: preserve their original
+ // behavior of falling back to GenericRecord rather than failing when the runtime type is
+ // missing from the classpath.
+ this.runtimeType = findClassOrFallbackToGeneric(userCodeClassLoader, runtimeTypeName);
this.runtimeSchema = tryExtractAvroSchema(userCodeClassLoader, runtimeType);
}
@@ -114,18 +121,22 @@ private void readV2(DataInputView in, ClassLoader userCodeClassLoader) throws IO
final String previousRuntimeTypeName = in.readUTF();
final String previousSchemaDefinition = in.readUTF();
- this.runtimeType = findClassOrThrow(userCodeClassLoader, previousRuntimeTypeName);
+ this.runtimeTypeName = previousRuntimeTypeName;
+ this.runtimeType = tryFindClass(userCodeClassLoader, runtimeTypeName);
this.schema = parseAvroSchema(previousSchemaDefinition);
- this.runtimeSchema = tryExtractAvroSchema(userCodeClassLoader, runtimeType);
+ this.runtimeSchema =
+ runtimeType == null ? null : tryExtractAvroSchema(userCodeClassLoader, runtimeType);
}
private void readV3(DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
final String previousRuntimeTypeName = readString(in);
final String previousSchemaDefinition = readString(in);
- this.runtimeType = findClassOrThrow(userCodeClassLoader, previousRuntimeTypeName);
+ this.runtimeTypeName = previousRuntimeTypeName;
+ this.runtimeType = tryFindClass(userCodeClassLoader, runtimeTypeName);
this.schema = parseAvroSchema(previousSchemaDefinition);
- this.runtimeSchema = tryExtractAvroSchema(userCodeClassLoader, runtimeType);
+ this.runtimeSchema =
+ runtimeType == null ? null : tryExtractAvroSchema(userCodeClassLoader, runtimeType);
}
@Override
@@ -141,9 +152,12 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility(
@Override
public TypeSerializer restoreSerializer() {
- checkNotNull(runtimeType);
checkNotNull(schema);
+ if (runtimeType == null) {
+ return CustomRestoreSerializerFactory.restoreFallbackSerializer(this, runtimeTypeName);
+ }
+
if (runtimeSchema != null) {
return new AvroSerializer<>(
runtimeType,
@@ -157,6 +171,11 @@ public TypeSerializer restoreSerializer() {
}
}
+ /** Returns the Avro writer schema stored in this snapshot. */
+ public Schema getSchema() {
+ return schema;
+ }
+
// ------------------------------------------------------------------------------------------------------------
// Helpers
// ------------------------------------------------------------------------------------------------------------
@@ -225,32 +244,16 @@ private static Schema tryExtractAvroSchema(ClassLoader cl, Class> runtimeType)
return d.getSchema(runtimeType);
}
- @SuppressWarnings("unchecked")
- @Nonnull
- private static Class findClassOrThrow(
- ClassLoader userCodeClassLoader, String className) {
- try {
- Class> runtimeTarget = Class.forName(className, false, userCodeClassLoader);
- return (Class) runtimeTarget;
- } catch (ClassNotFoundException e) {
- throw new IllegalStateException(
- ""
- + "Unable to find the class '"
- + className
- + "' which is used to deserialize "
- + "the elements of this serializer. "
- + "Were the class was moved or renamed?",
- e);
- }
+ @Nullable
+ private static Class tryFindClass(ClassLoader userCodeClassLoader, String className) {
+ return CustomRestoreSerializerFactory.resolveOrNull(className, userCodeClassLoader);
}
@SuppressWarnings("unchecked")
- @Nonnull
private static Class findClassOrFallbackToGeneric(
ClassLoader userCodeClassLoader, String className) {
try {
- Class> runtimeTarget = Class.forName(className, false, userCodeClassLoader);
- return (Class) runtimeTarget;
+ return (Class) Class.forName(className, false, userCodeClassLoader);
} catch (ClassNotFoundException e) {
return (Class) GenericRecord.class;
}
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshotTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshotTest.java
index c0838e9b842f9c..ba7b0231d806d1 100644
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshotTest.java
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerSnapshotTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.formats.avro.typeutils;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil;
@@ -45,6 +46,7 @@
import static org.apache.flink.api.common.typeutils.TypeSerializerConditions.isCompatibleAsIs;
import static org.apache.flink.api.common.typeutils.TypeSerializerConditions.isIncompatible;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test {@link AvroSerializerSnapshot}. */
class AvroSerializerSnapshotTest {
@@ -269,6 +271,75 @@ void restorePastSnapshots() throws IOException {
}
}
+ /**
+ * V1 snapshots predate {@code CustomRestoreSerializerFactory} support and must keep their
+ * original behavior of falling back to {@link GenericRecord} rather than failing when the
+ * runtime type is missing from the classpath (regression test for a fallback that was
+ * accidentally dropped while adding lenient reading for the State Processing API).
+ */
+ @Test
+ void restoringV1SnapshotWithMissingRuntimeTypeFallsBackToGenericRecord() throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ out.writeUTF(Address.getClassSchema().toString(false));
+
+ AvroSerializerSnapshot restored = new AvroSerializerSnapshot<>();
+ ClassLoader withoutAddress = classLoaderHiding(Address.class.getSimpleName());
+ restored.readSnapshot(1, new DataInputDeserializer(out.getCopyOfBuffer()), withoutAddress);
+
+ @SuppressWarnings("unchecked")
+ AvroSerializer serializer = (AvroSerializer) restored.restoreSerializer();
+ assertThat(serializer.getType()).isEqualTo(GenericRecord.class);
+ }
+
+ /**
+ * Regular job restores (i.e. without a {@link CustomRestoreSerializerFactory} registered, as is
+ * always the case outside of the State Processing API) must still fail fast when a V2/V3
+ * snapshot's runtime type is genuinely missing.
+ */
+ @Test
+ void restoringV3SnapshotWithMissingRuntimeTypeFailsFastWithoutFallbackFactory()
+ throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ new AvroSerializerSnapshot<>(Address.getClassSchema(), Address.class).writeSnapshot(out);
+ ClassLoader withoutAddress = classLoaderHiding(Address.class.getSimpleName());
+
+ assertThatThrownBy(
+ () ->
+ new AvroSerializerSnapshot()
+ .readSnapshot(
+ 3,
+ new DataInputDeserializer(out.getCopyOfBuffer()),
+ withoutAddress))
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasCauseInstanceOf(ClassNotFoundException.class);
+ }
+
+ /**
+ * With a {@link CustomRestoreSerializerFactory} registered, a V2/V3 snapshot must be readable
+ * even when the runtime type is missing from the classpath.
+ */
+ @Test
+ void restoringV3SnapshotWithMissingRuntimeTypeReadsLenientlyWithFallbackFactory()
+ throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ new AvroSerializerSnapshot<>(Address.getClassSchema(), Address.class).writeSnapshot(out);
+ ClassLoader withoutAddress = classLoaderHiding(Address.class.getSimpleName());
+
+ AvroSerializerSnapshot restored = new AvroSerializerSnapshot<>();
+ CustomRestoreSerializerFactory.set(
+ snapshot -> {
+ throw new UnsupportedOperationException("not exercised in this test");
+ });
+ try {
+ restored.readSnapshot(
+ 3, new DataInputDeserializer(out.getCopyOfBuffer()), withoutAddress);
+ } finally {
+ CustomRestoreSerializerFactory.remove();
+ }
+
+ assertThat(restored.getSchema()).isEqualTo(Address.getClassSchema());
+ }
+
/**
* Creates a new serializer snapshot for the current version. Use this before bumping the
* snapshot version and also add the version (before bumping) to {@link #PAST_VERSIONS}.
@@ -332,6 +403,20 @@ private static T deserialize(TypeSerializer serializer, ByteBuffer serial
return serializer.deserialize(in);
}
+ /** Returns a class loader that fails to find any class whose simple name is {@code hidden}. */
+ private static ClassLoader classLoaderHiding(String hidden) {
+ return new ClassLoader(AvroSerializerSnapshotTest.class.getClassLoader()) {
+ @Override
+ protected Class> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.contains(hidden)) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+ }
+
// ---------------------------------------------------------------------------------------------------------------
// Test classes
// ---------------------------------------------------------------------------------------------------------------
diff --git a/flink-libraries/flink-state-processing-api/pom.xml b/flink-libraries/flink-state-processing-api/pom.xml
index eb027221a0c4f1..d8cc8d1654d0d9 100644
--- a/flink-libraries/flink-state-processing-api/pom.xml
+++ b/flink-libraries/flink-state-processing-api/pom.xml
@@ -68,6 +68,13 @@ under the License.
provided
+
+ org.apache.flink
+ flink-table-type-utils
+ ${project.version}
+ provided
+
+
@@ -89,7 +96,7 @@ under the License.
org.apache.flinkflink-avro${project.version}
- test
+ true
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 00000000000000..b6a4eacba30b34
--- /dev/null
+++ b/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.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.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 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 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 getKeyedStates(
+ CheckpointMetadata metadata, OperatorIdentifier operatorId) throws IOException {
+ OperatorState opState = findOperatorState(metadata, operatorId);
+ List 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.
+ *
+ *
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 schemas = StateSchemaExtractor.extractSchema(opState);
+ ClassifiedStates classified = classifyStates(operatorId.toString(), schemas);
+ return buildKeyedStateSchemaInfo(schemas, classified.voidNamespaceStates, null);
+ }
+
+ private static KeyedStateSchemaInfo buildKeyedStateSchemaInfo(
+ List allSchemas,
+ List statesToInclude,
+ @Nullable LogicalType windowLogicalType) {
+ LogicalType keyType =
+ allSchemas.isEmpty()
+ ? new VarBinaryType(true, VarBinaryType.MAX_LENGTH)
+ : SerializerSnapshotToLogicalTypeConverter.convert(
+ allSchemas.get(0).keySnapshot);
+
+ LinkedHashMap 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.
+ *
+ *
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}.
+ *
+ *
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 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 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.
+ *
+ *
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:
+ *
+ *
+ *
+ * @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 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 buildBaseConnectorOptions(
+ String statePath, OperatorIdentifier operatorIdentifier) {
+ Map 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 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}.
+ *
+ *
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 detectStateBackendType(OperatorState opState) {
+ Set 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 handles, Set 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).
+ *
+ *
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 voidNamespaceStates;
+ final List windowNamespaceStates;
+ @Nullable final LogicalType windowLogicalType;
+
+ ClassifiedStates(
+ List voidNamespaceStates,
+ List 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.
+ *
+ *
An operator may register states under more than one distinct namespace type 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 schemas) {
+ List voidStates = new ArrayList<>();
+ Map> 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 chosenGroup = Collections.emptyList();
+ LogicalType chosenNamespaceType = null;
+ for (Map.Entry> 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 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/input/KeyedStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java
index 00bb5c262e1931..3850265a861829 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java
@@ -24,7 +24,8 @@
import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
import org.apache.flink.api.common.io.RichInputFormat;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.io.InputSplitAssigner;
@@ -38,6 +39,7 @@
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.state.api.filter.SavepointKeyFilter;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
+import org.apache.flink.state.api.input.deserializer.MissingClassSerializerFactory;
import org.apache.flink.state.api.input.operator.StateReaderOperator;
import org.apache.flink.state.api.input.splits.KeyGroupRangeInputSplit;
import org.apache.flink.state.api.runtime.SavepointRuntimeContext;
@@ -92,7 +94,7 @@ public class KeyedStateInputFormat
private transient BufferingCollector out;
- private transient CloseableIterator> keysAndNamespaces;
+ private transient CloseableIterator> keysAndNamespaces;
/**
* Creates an input format for reading partitioned state from an operator in a savepoint.
@@ -211,15 +213,10 @@ public void open(KeyGroupRangeInputSplit split) throws IOException {
registry = new CloseableRegistry();
RuntimeContext runtimeContext = getRuntimeContext();
- ExecutionConfig executionConfig;
- try {
- executionConfig =
- serializedExecutionConfig.deserializeValue(
- runtimeContext.getUserCodeClassLoader());
- } catch (ClassNotFoundException e) {
- throw new RuntimeException("Could not deserialize ExecutionConfig.", e);
- }
- final StreamOperatorStateContext context =
+ ExecutionConfig executionConfig =
+ deserialize(serializedExecutionConfig, runtimeContext.getUserCodeClassLoader());
+
+ StreamOperatorContextBuilder builder =
new StreamOperatorContextBuilder(
runtimeContext,
configuration,
@@ -229,19 +226,29 @@ 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()));
- AbstractKeyedStateBackend keyedStateBackend =
- (AbstractKeyedStateBackend) context.keyedStateBackend();
+ // 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.
+ CustomRestoreSerializerFactory.set(MissingClassSerializerFactory::create);
- final DefaultKeyedStateStore keyedStateStore =
- new DefaultKeyedStateStore(keyedStateBackend, runtimeContext::createSerializer);
- SavepointRuntimeContext ctx = new SavepointRuntimeContext(runtimeContext, keyedStateStore);
-
- InternalTimeServiceManager timeServiceManager =
- (InternalTimeServiceManager) context.internalTimerServiceManager();
try {
+ final StreamOperatorStateContext context = builder.build(LOG);
+
+ AbstractKeyedStateBackend keyedStateBackend =
+ (AbstractKeyedStateBackend) context.keyedStateBackend();
+
+ final DefaultKeyedStateStore keyedStateStore =
+ new DefaultKeyedStateStore(keyedStateBackend, runtimeContext::createSerializer);
+ SavepointRuntimeContext ctx =
+ new SavepointRuntimeContext(runtimeContext, keyedStateStore);
+
+ InternalTimeServiceManager timeServiceManager =
+ (InternalTimeServiceManager) context.internalTimerServiceManager();
+
operator.setup(
runtimeContext::createSerializer, keyedStateBackend, timeServiceManager, ctx);
operator.open();
@@ -280,8 +287,8 @@ public OUT nextRecord(OUT reuse) throws IOException {
return out.next();
}
- final Tuple2 keyAndNamespace = keysAndNamespaces.next();
- operator.setCurrentKey(keyAndNamespace.f0);
+ final Tuple3 keyAndNamespace = keysAndNamespaces.next();
+ operator.setCurrentKeyAndKeyGroup(keyAndNamespace.f0, keyAndNamespace.f2);
try {
operator.processElement(keyAndNamespace.f0, keyAndNamespace.f1, out);
@@ -325,4 +332,14 @@ private static List sortedKeyGroupRanges(int minNumSplits, int ma
keyGroups.sort(Comparator.comparing(KeyGroupRange::getStartKeyGroup));
return keyGroups;
}
+
+ static ExecutionConfig deserialize(
+ SerializedValue serializedExecutionConfig, ClassLoader classLoader)
+ throws IOException {
+ try {
+ return serializedExecutionConfig.deserializeValue(classLoader);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Could not deserialize ExecutionConfig.", e);
+ }
+ }
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java
index 38d1d3505a2577..7d6953eac51bbe 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java
@@ -20,6 +20,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.VoidNamespace;
@@ -36,13 +37,20 @@
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
+ *
Uses {@link KeyedStateBackend#getKeysAndKeyGroups} rather than {@link
+ * KeyedStateBackend#getKeys(List, Object)}, because the State Processing API can read state whose
+ * key class is missing from the classpath, deserializing the key into a substitute representation
+ * instead of failing (see {@code CustomRestoreSerializerFactory}). Such a substituted key's {@code
+ * hashCode()} is not guaranteed to reproduce the original, so the key-group it was physically
+ * written under must be carried alongside it rather than recomputed from the key.
+ *
* @param Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator implements CloseableIterator {
+public final class MultiStateKeyIterator implements CloseableIterator> {
private final List extends StateDescriptor, ?>> descriptors;
- private final Iterator iterator;
+ private final Iterator> iterator;
private final CloseableRegistry registry;
@@ -52,8 +60,8 @@ public MultiStateKeyIterator(
this.descriptors = Preconditions.checkNotNull(descriptors);
Preconditions.checkNotNull(backend);
registry = new CloseableRegistry();
- Stream stream =
- backend.getKeys(
+ Stream> stream =
+ backend.getKeysAndKeyGroups(
this.descriptors.stream()
.map(StateDescriptor::getName)
.collect(Collectors.toList()),
@@ -72,7 +80,7 @@ public boolean hasNext() {
}
@Override
- public K next() {
+ public Tuple2 next() {
if (!hasNext()) {
throw new NoSuchElementException();
} else {
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java
index 978160dc0e069a..65745a1b16ab55 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java
@@ -173,14 +173,9 @@ public void open(OperatorStateInputSplit split) throws IOException {
registry = new CloseableRegistry();
RuntimeContext runtimeContext = getRuntimeContext();
- ExecutionConfig executionConfig;
- try {
- executionConfig =
- serializedExecutionConfig.deserializeValue(
- runtimeContext.getUserCodeClassLoader());
- } catch (ClassNotFoundException e) {
- throw new RuntimeException("Could not deserialize ExecutionConfig.", e);
- }
+ ExecutionConfig executionConfig =
+ KeyedStateInputFormat.deserialize(
+ serializedExecutionConfig, runtimeContext.getUserCodeClassLoader());
final StreamOperatorStateContext context =
new StreamOperatorContextBuilder(
runtimeContext,
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/EnumNameDeserializer.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/EnumNameDeserializer.java
new file mode 100644
index 00000000000000..5105b1d65a048e
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/EnumNameDeserializer.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.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.common.typeutils.base.EnumSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * A {@link TypeSerializer} that reads the enum ordinal written by {@link EnumSerializer} and
+ * produces the enum constant's {@code name()} as a plain {@link String}, without the user enum
+ * class being on the classpath.
+ *
+ *
{@link EnumSerializer} writes a maintained ordinal (see {@link
+ * EnumSerializer.EnumSerializerSnapshot#getEnumNames()}) rather than {@link Enum#ordinal()}, so the
+ * name lookup here uses the exact same array the original serializer would have used to resolve
+ * that ordinal back to a constant.
+ */
+@Internal
+public final class EnumNameDeserializer extends TypeSerializer {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String[] enumNames;
+
+ public static EnumNameDeserializer create(EnumSerializer.EnumSerializerSnapshot> snapshot) {
+ return new EnumNameDeserializer(snapshot.getEnumNames());
+ }
+
+ EnumNameDeserializer(String[] enumNames) {
+ this.enumNames = enumNames;
+ }
+
+ @Override
+ public String deserialize(DataInputView source) throws IOException {
+ int ordinal = source.readInt();
+ if (ordinal < 0 || ordinal >= enumNames.length) {
+ throw new IOException(
+ "Unknown enum ordinal "
+ + ordinal
+ + " (have "
+ + enumNames.length
+ + " known constants). The savepoint may have been written with a"
+ + " different enum definition.");
+ }
+ return enumNames[ordinal];
+ }
+
+ @Override
+ public String deserialize(String reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ // -------------------------------------------------------------------------
+ // TypeSerializer boilerplate — copy/snapshot operations not needed for
+ // schema-extraction use cases but required by the interface.
+ // -------------------------------------------------------------------------
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public TypeSerializer duplicate() {
+ return this;
+ }
+
+ @Override
+ public String createInstance() {
+ return enumNames.length > 0 ? enumNames[0] : null;
+ }
+
+ @Override
+ public String copy(String from) {
+ return from;
+ }
+
+ @Override
+ public String copy(String from, String reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return 4;
+ }
+
+ @Override
+ public void serialize(String record, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "EnumNameDeserializer is read-only; serialization is not supported.");
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "EnumNameDeserializer is read-only; copy is not supported.");
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof EnumNameDeserializer)) {
+ return false;
+ }
+ return Arrays.equals(enumNames, ((EnumNameDeserializer) obj).enumNames);
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(enumNames);
+ }
+
+ @Override
+ public TypeSerializerSnapshot snapshotConfiguration() {
+ throw new UnsupportedOperationException(
+ "EnumNameDeserializer is only ever used directly within a single read; it is never"
+ + " re-snapshotted.");
+ }
+}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverter.java
new file mode 100644
index 00000000000000..2d726f3de26a6f
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverter.java
@@ -0,0 +1,293 @@
+/*
+ * 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}.
+ *
+ *