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/java/typeutils/runtime/PojoDeserializerCompatibilitySnapshot.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoDeserializerCompatibilitySnapshot.java
new file mode 100644
index 00000000000000..0dace3f1537fd1
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoDeserializerCompatibilitySnapshot.java
@@ -0,0 +1,93 @@
+/*
+ * 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 java.io.IOException;
+
+/**
+ * A {@link TypeSerializerSnapshot} for deserializers that can read POJO binary data without the
+ * user POJO class being on the classpath.
+ *
+ *
This snapshot declares itself {@link TypeSerializerSchemaCompatibility#compatibleAsIs()
+ * compatible as-is} with any stored {@link PojoSerializerSnapshot}. It is used by the {@code
+ * PojoToRowDataDeserializer} in the state processing API (which converts them to {@code
+ * GenericRowData}).
+ *
+ *
This snapshot only ever exists in-memory, wrapping the live deserializer it was created from:
+ * composite schema compatibility checks (e.g. {@code
+ * CompositeTypeSerializerSnapshot#resolveOuterSchemaCompatibility}) restore the "new" side of a
+ * composite serializer (such as a timer serializer nesting the key serializer) to compare it
+ * against the old one, even when nested-level compatibility already short-circuited to {@link
+ * TypeSerializerSchemaCompatibility#compatibleAsIs()}. {@link #restoreSerializer()} therefore
+ * returns the wrapped instance instead of throwing, since it never actually needs to be
+ * reconstructed from a persisted snapshot.
+ */
+@Internal
+public final class PojoDeserializerCompatibilitySnapshot implements TypeSerializerSnapshot {
+
+ private final TypeSerializer restoredSerializer;
+
+ public PojoDeserializerCompatibilitySnapshot() {
+ this.restoredSerializer = null;
+ }
+
+ public PojoDeserializerCompatibilitySnapshot(TypeSerializer restoredSerializer) {
+ this.restoredSerializer = restoredSerializer;
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility(
+ TypeSerializerSnapshot oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof PojoSerializerSnapshot
+ || oldSerializerSnapshot instanceof PojoDeserializerCompatibilitySnapshot) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ @Override
+ public TypeSerializer restoreSerializer() {
+ if (restoredSerializer != null) {
+ return restoredSerializer;
+ }
+ throw new UnsupportedOperationException(
+ "PojoDeserializerCompatibilitySnapshot cannot reconstruct the deserializer on its own. "
+ + "Use PojoSerializerSnapshot.restoreSerializer() or "
+ + "PojoToRowDataDeserializer.create().");
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {}
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+ throws IOException {}
+}
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..bb78d725f8aca2 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
@@ -54,6 +54,42 @@ public class PojoSerializerSnapshot implements TypeSerializerSnapshot {
*/
private static final int VERSION = 2;
+ /**
+ * Optional thread-local factory for building a custom serializer when the POJO class is not on
+ * the classpath.
+ *
+ *
Used by the State Processing API to inject {@code PojoToRowDataDeserializer} so that the
+ * heap state backend can restore POJO state without the user class (values are converted to
+ * {@code GenericRowData} during restore instead of raising {@link ClassNotFoundException}).
+ *
+ *
Set via {@link #setPojoRestoreSerializerFactory} before the state backend restore and
+ * cleared via {@link #clearPojoRestoreSerializerFactory} afterward.
+ */
+ private static final ThreadLocal<
+ java.util.function.Function, TypeSerializer>>>
+ RESTORE_FACTORY = new ThreadLocal<>();
+
+ /**
+ * Registers a factory that provides a fallback {@link TypeSerializer} when the POJO class is
+ * absent from the classpath.
+ *
+ *
The factory receives the snapshot and should return a serializer that can read the POJO
+ * binary format.
+ *
+ *
Must be paired with {@link #clearPojoRestoreSerializerFactory()} in a try-finally block.
+ */
+ @Internal
+ public static void setPojoRestoreSerializerFactory(
+ java.util.function.Function, TypeSerializer>> factory) {
+ RESTORE_FACTORY.set(factory);
+ }
+
+ /** Clears the factory registered via {@link #setPojoRestoreSerializerFactory}. */
+ @Internal
+ public static void clearPojoRestoreSerializerFactory() {
+ RESTORE_FACTORY.remove();
+ }
+
/** Contains the actual content for the serializer snapshot. */
private PojoSerializerSnapshotData snapshotData;
@@ -143,6 +179,15 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
@Override
@SuppressWarnings("unchecked")
public TypeSerializer restoreSerializer() {
+ if (snapshotData.getPojoClass() == null) {
+ java.util.function.Function, TypeSerializer>> factory =
+ RESTORE_FACTORY.get();
+ if (factory != null) {
+ return (TypeSerializer) factory.apply(this);
+ }
+ throw new RuntimeException(new ClassNotFoundException(snapshotData.getPojoClassName()));
+ }
+
final int numFields = snapshotData.getFieldSerializerSnapshots().size();
final ArrayList restoredFields = new ArrayList<>(numFields);
@@ -257,6 +302,73 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility(
return TypeSerializerSchemaCompatibility.compatibleAsIs();
}
+ // ---------------------------------------------------------------------------------------------
+ // Schema extraction support
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the raw snapshot data. Exposed for schema extraction utilities that need to inspect
+ * field names and field serializer snapshots without the user POJO class being on the
+ * classpath.
+ *
+ * @return the snapshot data; field names are always present, field serializer snapshots may be
+ * null for individual fields if their snapshot could not be read
+ */
+ @Internal
+ public PojoSerializerSnapshotData getSnapshotData() {
+ return snapshotData;
+ }
+
+ /**
+ * Returns {@code true} if the POJO class can be loaded from the current classloader.
+ *
+ *
When {@code false}, {@link #restoreSerializer()} returns a (or a factory-supplied
+ * deserializer) instead of the normal {@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 TypeSerializerSnapshot) pairs. Field names are
+ * always present; snapshot values may be {@code null} when the field snapshot could not be
+ * read.
+ */
+ @Internal
+ public java.util.List>>
+ getFieldSnapshotEntries() {
+ java.util.List>>
+ result = new java.util.ArrayList<>();
+ snapshotData
+ .getFieldSerializerSnapshots()
+ .forEach(
+ (fieldName, field, fieldSnapshot) ->
+ result.add(
+ new java.util.AbstractMap.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 java.util.List> getRegisteredSubclassSnapshotsOrdered() {
+ return new java.util.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..12a3b58807f1f5 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
@@ -24,7 +24,6 @@
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 +31,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 +114,7 @@ static PojoSerializerSnapshotData createFrom(
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClass.getName(),
fieldSerializerSnapshots,
optionalMapOf(registeredSubclassSerializerSnapshots, Class::getName),
optionalMapOf(nonRegisteredSubclassSerializerSnapshots, Class::getName));
@@ -153,12 +155,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 +170,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 +192,7 @@ private PojoSerializerSnapshotData(
// ---------------------------------------------------------------------------------------------
void writeSnapshotData(DataOutputView out) throws IOException {
- out.writeUTF(pojoClass.getName());
+ out.writeUTF(pojoClassName);
writeOptionalMap(
out,
fieldSerializerSnapshots,
@@ -206,7 +212,17 @@ 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 = null;
+ try {
+ @SuppressWarnings("unchecked")
+ Class resolved = (Class) Class.forName(pojoClassName, false, userCodeClassLoader);
+ pojoClass = resolved;
+ } catch (ClassNotFoundException e) {
+ LOG.debug(
+ "POJO class '{}' not found on classpath; schema can still be read from field snapshots.",
+ pojoClassName);
+ }
LinkedOptionalMap> fieldSerializerSnapshots =
readOptionalMap(
@@ -226,6 +242,7 @@ private static PojoSerializerSnapshotData readSnapshotData(
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClassName,
fieldSerializerSnapshots,
registeredSubclassSerializerSnapshots,
nonRegisteredSubclassSerializerSnapshots);
@@ -235,10 +252,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/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..f2789979f8e7db
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshotLenientReadTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.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.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.List;
+
+/**
+ * Tests for the lenient POJO class-loading fix in {@link PojoSerializerSnapshotData}.
+ *
+ *
Verifies that a {@link PojoSerializerSnapshot} can be read back even when the POJO class is
+ * not on the classpath — field names and field serializer snapshots must remain accessible.
+ */
+public class PojoSerializerSnapshotLenientReadTest {
+
+ // -------------------------------------------------------------------------
+ // Simple POJO used for writing snapshots (available during write, absent during read)
+ // -------------------------------------------------------------------------
+
+ public static class SomePojo {
+ public String name;
+ public int age;
+ public long score;
+ }
+
+ // -------------------------------------------------------------------------
+ // Tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ public void testReadWithClassPresent() throws IOException {
+ PojoSerializerSnapshot written = buildSnapshot(SomePojo.class);
+ PojoSerializerSnapshot read =
+ roundtripSnapshot(written, getClass().getClassLoader());
+
+ // Class should be found normally.
+ Assert.assertNotNull(read.getSnapshotData().getPojoClass());
+ Assert.assertTrue(
+ "Class name should contain SomePojo",
+ read.getSnapshotData().getPojoClassName().contains("SomePojo"));
+
+ List>> entries =
+ read.getFieldSnapshotEntries();
+ Assert.assertEquals(3, entries.size());
+ assertFieldEntry(entries, "name", StringSerializer.StringSerializerSnapshot.class);
+ assertFieldEntry(entries, "age", IntSerializer.IntSerializerSnapshot.class);
+ assertFieldEntry(entries, "score", LongSerializer.LongSerializerSnapshot.class);
+ }
+
+ @Test
+ public void testReadWithClassAbsent() throws IOException {
+ PojoSerializerSnapshot written = buildSnapshot(SomePojo.class);
+
+ // Use a classloader that cannot see SomePojo.
+ ClassLoader noPojoClassLoader =
+ new ClassLoader(getClass().getClassLoader()) {
+ @Override
+ public Class> loadClass(String name) throws ClassNotFoundException {
+ if (name.equals(SomePojo.class.getName())) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name);
+ }
+ };
+
+ PojoSerializerSnapshot> read = roundtripSnapshot(written, noPojoClassLoader);
+
+ // Class should be null but className should be preserved.
+ Assert.assertNull(read.getSnapshotData().getPojoClass());
+ Assert.assertTrue(read.getPojoClassName().endsWith("SomePojo"));
+
+ // Field names and snapshots must still be readable.
+ List>> entries =
+ read.getFieldSnapshotEntries();
+ Assert.assertEquals(3, entries.size());
+ assertFieldEntry(entries, "name", StringSerializer.StringSerializerSnapshot.class);
+ assertFieldEntry(entries, "age", IntSerializer.IntSerializerSnapshot.class);
+ assertFieldEntry(entries, "score", LongSerializer.LongSerializerSnapshot.class);
+ }
+
+ @Test
+ public void testGetPojoClassNameAlwaysSet() throws IOException {
+ PojoSerializerSnapshot written = buildSnapshot(SomePojo.class);
+ PojoSerializerSnapshot> read = roundtripSnapshot(written, getClass().getClassLoader());
+
+ // getPojoClassName must always return the class name, regardless of loading.
+ Assert.assertEquals(SomePojo.class.getName(), read.getPojoClassName());
+ }
+
+ @Test
+ public void testFieldNamePreservedWhenFieldClassAbsent() throws IOException {
+ // Write a snapshot for a POJO that has a field referencing another POJO class.
+ // When read back without the field's declaring class, the field name must still
+ // appear in the entries (key name is always written before the framed value).
+ PojoSerializerSnapshot written = buildSnapshot(SomePojo.class);
+
+ ClassLoader noFieldLoader =
+ new ClassLoader(getClass().getClassLoader()) {
+ @Override
+ protected Class> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.contains("SomePojo")) {
+ throw new ClassNotFoundException(name);
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+
+ PojoSerializerSnapshot> read = roundtripSnapshot(written, noFieldLoader);
+
+ List>> entries =
+ read.getFieldSnapshotEntries();
+ Assert.assertEquals(3, entries.size());
+ // Names must be present even when class is absent (order may vary).
+ Assert.assertTrue(entries.stream().anyMatch(e -> "name".equals(e.getKey())));
+ Assert.assertTrue(entries.stream().anyMatch(e -> "age".equals(e.getKey())));
+ Assert.assertTrue(entries.stream().anyMatch(e -> "score".equals(e.getKey())));
+ }
+
+ // -------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static PojoSerializerSnapshot buildSnapshot(Class clazz) {
+ return (PojoSerializerSnapshot)
+ org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfo(clazz)
+ .createSerializer(new SerializerConfigImpl())
+ .snapshotConfiguration();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PojoSerializerSnapshot roundtripSnapshot(
+ PojoSerializerSnapshot snapshot, ClassLoader cl) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return (PojoSerializerSnapshot) TypeSerializerSnapshot.readVersionedSnapshot(in, cl);
+ }
+
+ private static void assertFieldEntry(
+ List>> entries,
+ String expectedName,
+ Class> expectedSnapshotType) {
+ for (AbstractMap.SimpleEntry> entry : entries) {
+ if (expectedName.equals(entry.getKey())) {
+ Assert.assertNotNull(
+ "Snapshot for field '" + expectedName + "' should not be null",
+ entry.getValue());
+ Assert.assertSame(
+ "Wrong snapshot type for field '" + expectedName + "'",
+ expectedSnapshotType,
+ entry.getValue().getClass());
+ return;
+ }
+ }
+ Assert.fail("Field '" + expectedName + "' not found in snapshot entries");
+ }
+}
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..c4c8b622d1bab5 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
@@ -32,6 +32,8 @@
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.specific.SpecificData;
import org.apache.avro.specific.SpecificRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
@@ -49,6 +51,7 @@
* @param The data type that the originating serializer of this configuration serializes.
*/
public class AvroSerializerSnapshot implements TypeSerializerSnapshot {
+ private static final Logger LOG = LoggerFactory.getLogger(AvroSerializerSnapshot.class);
private Class runtimeType;
private Schema schema;
private Schema runtimeSchema;
@@ -114,7 +117,8 @@ 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.runtimeType =
+ findClassOrFallbackToGeneric(userCodeClassLoader, previousRuntimeTypeName);
this.schema = parseAvroSchema(previousSchemaDefinition);
this.runtimeSchema = tryExtractAvroSchema(userCodeClassLoader, runtimeType);
}
@@ -123,7 +127,8 @@ private void readV3(DataInputView in, ClassLoader userCodeClassLoader) throws IO
final String previousRuntimeTypeName = readString(in);
final String previousSchemaDefinition = readString(in);
- this.runtimeType = findClassOrThrow(userCodeClassLoader, previousRuntimeTypeName);
+ this.runtimeType =
+ findClassOrFallbackToGeneric(userCodeClassLoader, previousRuntimeTypeName);
this.schema = parseAvroSchema(previousSchemaDefinition);
this.runtimeSchema = tryExtractAvroSchema(userCodeClassLoader, runtimeType);
}
@@ -161,6 +166,11 @@ public TypeSerializer restoreSerializer() {
// Helpers
// ------------------------------------------------------------------------------------------------------------
+ /** Returns the Avro writer schema stored in this snapshot. */
+ public Schema getSchema() {
+ return schema;
+ }
+
/**
* Resolves writer/reader schema compatibly.
*
@@ -252,6 +262,9 @@ private static Class findClassOrFallbackToGeneric(
Class> runtimeTarget = Class.forName(className, false, userCodeClassLoader);
return (Class) runtimeTarget;
} catch (ClassNotFoundException e) {
+ LOG.debug(
+ "Avro runtime type '{}' not found on classpath; falling back to GenericRecord.",
+ className);
return (Class) GenericRecord.class;
}
}
diff --git a/flink-libraries/flink-state-processing-api/pom.xml b/flink-libraries/flink-state-processing-api/pom.xml
index eb027221a0c4f1..b9fff64d9a73bf 100644
--- a/flink-libraries/flink-state-processing-api/pom.xml
+++ b/flink-libraries/flink-state-processing-api/pom.xml
@@ -89,7 +89,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..0d2e269686bcaf
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateTableUtils.java
@@ -0,0 +1,635 @@
+/*
+ * 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 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.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 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.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.
+ *
+ *
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);
+ 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) {
+ // Flink's built-in timer states use this prefix.
+ // "merging-window-set" is WindowOperator's internal session-window merging bookkeeping
+ // state, not a user-registered state.
+ return stateName.startsWith("_timer_state/") || stateName.equals("merging-window-set");
+ }
+
+ /**
+ * 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 (Exception 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/ExecutionConfigs.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/ExecutionConfigs.java
new file mode 100644
index 00000000000000..679a8bf1ae8637
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/ExecutionConfigs.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.util.SerializedValue;
+
+import java.io.IOException;
+
+/**
+ * Shared helper for deserializing the {@link ExecutionConfig} that {@link KeyedStateInputFormat}
+ * and {@link OperatorStateInputFormat} ship to task managers as a {@link SerializedValue}.
+ */
+@Internal
+final class ExecutionConfigs {
+
+ private ExecutionConfigs() {}
+
+ 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/KeyedStateInputFormat.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java
index 00bb5c262e1931..3967812f064e1d 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.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
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.PojoToRowDataDeserializer;
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,11 @@ 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 =
+ ExecutionConfigs.deserialize(
+ serializedExecutionConfig, runtimeContext.getUserCodeClassLoader());
+
+ StreamOperatorContextBuilder builder =
new StreamOperatorContextBuilder(
runtimeContext,
configuration,
@@ -229,19 +227,35 @@ 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()));
+
+ // Inject PojoToRowDataDeserializer as the restore serializer for any PojoSerializerSnapshot
+ // whose POJO class is absent from the current classpath. PojoSerializerSnapshot only
+ // consults the factory when getPojoClass() == null, so setting it unconditionally is safe:
+ // snapshots whose class is present restore normally and never invoke the factory.
+ // The factory runs on this task thread, which is the same thread that drives all
+ // subsequent state backend access, so a ThreadLocal is safe here. The scope must stay
+ // active beyond open(): the heap backend resolves state schema compatibility eagerly
+ // during build(), but the RocksDB backend defers it to the first actual state access,
+ // which happens lazily while processing elements in nextRecord(). The factory is only
+ // cleared in close(), once no further state access can occur on this thread.
+ PojoSerializerSnapshot.setPojoRestoreSerializerFactory(
+ pojoSnapshot -> PojoToRowDataDeserializer.create(pojoSnapshot));
- AbstractKeyedStateBackend keyedStateBackend =
- (AbstractKeyedStateBackend) context.keyedStateBackend();
+ try {
+ final StreamOperatorStateContext context = builder.build(LOG);
- final DefaultKeyedStateStore keyedStateStore =
- new DefaultKeyedStateStore(keyedStateBackend, runtimeContext::createSerializer);
- SavepointRuntimeContext ctx = new SavepointRuntimeContext(runtimeContext, keyedStateStore);
+ AbstractKeyedStateBackend keyedStateBackend =
+ (AbstractKeyedStateBackend) context.keyedStateBackend();
+
+ final DefaultKeyedStateStore keyedStateStore =
+ new DefaultKeyedStateStore(keyedStateBackend, runtimeContext::createSerializer);
+ SavepointRuntimeContext ctx =
+ new SavepointRuntimeContext(runtimeContext, keyedStateStore);
+
+ InternalTimeServiceManager timeServiceManager =
+ (InternalTimeServiceManager) context.internalTimerServiceManager();
- InternalTimeServiceManager timeServiceManager =
- (InternalTimeServiceManager) context.internalTimerServiceManager();
- try {
operator.setup(
runtimeContext::createSerializer, keyedStateBackend, timeServiceManager, ctx);
operator.open();
@@ -254,6 +268,7 @@ public void open(KeyGroupRangeInputSplit split) throws IOException {
keysAndNamespaces = operator.getKeysAndNamespaces(ctx);
}
} catch (Exception e) {
+ PojoSerializerSnapshot.clearPojoRestoreSerializerFactory();
throw new IOException("Failed to restore timer state", e);
}
}
@@ -266,6 +281,8 @@ public void close() throws IOException {
IOUtils.closeQuietly(registry);
} catch (Exception e) {
throw new IOException("Failed to close state backend", e);
+ } finally {
+ PojoSerializerSnapshot.clearPojoRestoreSerializerFactory();
}
}
@@ -280,8 +297,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);
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..23650914d30a15 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;
@@ -39,10 +40,10 @@
* @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 +53,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 +73,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..6597e4f38cb1b0 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 =
+ ExecutionConfigs.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/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..8d4c942e7ce3b6
--- /dev/null
+++ b/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.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+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.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}.
+ *
+ *