You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/08/24 09:16:44 UTC
(flink) 03/06: [FLINK-40177][state-processor-api] Serializer and runtime changes for type inference
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3d3fad390559be0f3968b5a7100999f98e841017
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Sun Jul 19 15:10:08 2026 +0200
[FLINK-40177][state-processor-api] Serializer and runtime changes for type inference
Adds the serializer-level support needed to convert savepoint state into
Flink table rows without the original classes on the classpath:
- PojoSerializerSnapshot exposes its field serializer snapshots and
registered subclass snapshots, and PojoDeserializerCompatibilitySnapshot
lets a PojoToRowDataDeserializer round-trip through
snapshotConfiguration()/restoreSerializer().
- AvroSerializerSnapshot exposes its Avro schema for conversion to a
LogicalType.
- RowDataSerializerSnapshot exposes its stored types and field names
(getTypes()/getFieldNames()) for conversion to a LogicalType.
- PojoToRowDataDeserializer reads the POJO binary format produced by
PojoSerializer and produces GenericRowData directly, using
InternalTypeConverter to convert individual field values to their
table-internal representation.
flink-state-processing-api now depends on flink-avro as a compile-scope
(optional) dependency, since SerializerSnapshotToLogicalTypeConverter
references Avro serializer snapshot classes directly.
---
.../typeutils/CustomRestoreSerializerFactory.java | 118 ++++++++
.../api/common/typeutils/base/EnumSerializer.java | 36 ++-
.../api/java/typeutils/runtime/PojoSerializer.java | 8 +-
.../typeutils/runtime/PojoSerializerSnapshot.java | 58 ++++
.../runtime/PojoSerializerSnapshotData.java | 32 ++-
.../EnumSerializerSnapshotMissingClassTest.java | 104 +++++++
.../PojoSerializerSnapshotLenientReadTest.java | 173 ++++++++++++
.../avro/typeutils/AvroSerializerSnapshot.java | 59 ++--
.../avro/typeutils/AvroSerializerSnapshotTest.java | 85 ++++++
flink-libraries/flink-state-processing-api/pom.xml | 9 +-
.../input/deserializer/EnumNameDeserializer.java | 145 ++++++++++
.../input/deserializer/InternalTypeConverter.java | 293 +++++++++++++++++++
.../PojoDeserializerCompatibilitySnapshot.java | 89 ++++++
.../deserializer/PojoToRowDataDeserializer.java | 314 +++++++++++++++++++++
.../flink/state/api/KeyedStateReadingITCase.java | 172 +++++++++++
.../deserializer/InternalTypeConverterTest.java | 301 ++++++++++++++++++++
.../PojoToRowDataDeserializerTest.java | 239 ++++++++++++++++
.../table/runtime/typeutils/RowDataSerializer.java | 17 ++
18 files changed, 2205 insertions(+), 47 deletions(-)
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 00000000000..3ad8c0870f0
--- /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.
+ *
+ * <p><b>This exists solely to support the Flink State Processing API's ability to read state whose
+ * original classes are not on the classpath</b> (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.
+ *
+ * <p>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<Function<TypeSerializerSnapshot<?>, TypeSerializer<?>>>
+ FACTORY = new ThreadLocal<>();
+
+ private CustomRestoreSerializerFactory() {}
+
+ /** Registers the fallback factory for the current thread. */
+ public static void set(Function<TypeSerializerSnapshot<?>, TypeSerializer<?>> factory) {
+ FACTORY.set(factory);
+ }
+
+ /** Returns the fallback factory registered via {@link #set}, or {@code null} if none. */
+ public static Function<TypeSerializerSnapshot<?>, 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 <T> Class<T> resolveOrNull(String className, ClassLoader classLoader) {
+ try {
+ return (Class<T>) 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 <T> TypeSerializer<T> restoreFallbackSerializer(
+ TypeSerializerSnapshot<T> snapshot, String missingClassName) {
+ Function<TypeSerializerSnapshot<?>, TypeSerializer<?>> fallback = get();
+ if (fallback == null) {
+ throw missingClass(missingClassName, new ClassNotFoundException(missingClassName));
+ }
+ return (TypeSerializer<T>) 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 ca7d1fc5632..ee5b5898ab2 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 @@ package org.apache.flink.api.common.typeutils.base;
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 final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
private T[] enums;
private Class<T> enumClass;
+ private String enumClassName;
+ private String[] enumNames;
@SuppressWarnings("unused")
public EnumSerializerSnapshot() {
@@ -213,16 +215,23 @@ public final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
@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 final class EnumSerializer<T extends Enum<T>> extends TypeSerializer<T> {
+ ". 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<T> 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 9d7e4254e86..19e31c9f211 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 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public final class PojoSerializer<T> extends TypeSerializer<T> {
// 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 cb8deb5b9e7..cdfe1b9b12e 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.SerializerConfig;
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 org.apache.flink.util.LinkedOptionalMap;
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 class PojoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
@Override
@SuppressWarnings("unchecked")
public TypeSerializer<T> restoreSerializer() {
+ if (snapshotData.getPojoClass() == null) {
+ return CustomRestoreSerializerFactory.restoreFallbackSerializer(
+ this, snapshotData.getPojoClassName());
+ }
+
final int numFields = snapshotData.getFieldSerializerSnapshots().size();
final ArrayList<Field> restoredFields = new ArrayList<>(numFields);
@@ -257,6 +265,56 @@ public class PojoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
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<SimpleEntry<String, TypeSerializerSnapshot<?>>> getFieldSnapshotEntries() {
+ List<SimpleEntry<String, TypeSerializerSnapshot<?>>> 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<TypeSerializerSnapshot<?>> 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 6b2bd112ad1..ceaaa3158e4 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.apache.flink.util.function.BiFunctionWithException;
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 @@ final class PojoSerializerSnapshotData<T> {
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClass.getName(),
fieldSerializerSnapshots,
optionalMapOf(registeredSubclassSerializerSnapshots, Class::getName),
optionalMapOf(nonRegisteredSubclassSerializerSnapshots, Class::getName));
@@ -153,12 +156,14 @@ final class PojoSerializerSnapshotData<T> {
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClass.getName(),
fieldSerializerSnapshots,
optionalMapOf(existingRegisteredSubclassSerializerSnapshots, Class::getName),
optionalMapOf(existingNonRegisteredSubclassSerializerSnapshots, Class::getName));
}
- private Class<T> pojoClass;
+ @Nullable private Class<T> pojoClass;
+ private String pojoClassName;
private LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots;
private LinkedOptionalMap<Class<?>, TypeSerializerSnapshot<?>>
registeredSubclassSerializerSnapshots;
@@ -166,14 +171,16 @@ final class PojoSerializerSnapshotData<T> {
nonRegisteredSubclassSerializerSnapshots;
private PojoSerializerSnapshotData(
- Class<T> typeClass,
+ @Nullable Class<T> typeClass,
+ String pojoClassName,
LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots,
LinkedOptionalMap<Class<?>, TypeSerializerSnapshot<?>>
registeredSubclassSerializerSnapshots,
LinkedOptionalMap<Class<?>, 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 @@ final class PojoSerializerSnapshotData<T> {
// ---------------------------------------------------------------------------------------------
void writeSnapshotData(DataOutputView out) throws IOException {
- out.writeUTF(pojoClass.getName());
+ out.writeUTF(pojoClassName);
writeOptionalMap(
out,
fieldSerializerSnapshots,
@@ -206,7 +213,14 @@ final class PojoSerializerSnapshotData<T> {
private static <T> PojoSerializerSnapshotData<T> readSnapshotData(
DataInputView in, ClassLoader userCodeClassLoader) throws IOException {
- Class<T> pojoClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+ final String pojoClassName = in.readUTF();
+ Class<T> pojoClass =
+ 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<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots =
readOptionalMap(
@@ -226,6 +240,7 @@ final class PojoSerializerSnapshotData<T> {
return new PojoSerializerSnapshotData<>(
pojoClass,
+ pojoClassName,
fieldSerializerSnapshots,
registeredSubclassSerializerSnapshots,
nonRegisteredSubclassSerializerSnapshots);
@@ -235,10 +250,15 @@ final class PojoSerializerSnapshotData<T> {
// Snapshot data accessors
// ---------------------------------------------------------------------------------------------
+ @Nullable
Class<T> getPojoClass() {
return pojoClass;
}
+ String getPojoClassName() {
+ return pojoClassName;
+ }
+
LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> 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 00000000000..a485c7cf85b
--- /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<SomeEnum> 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<SomeEnum> writeSnapshot() {
+ return new EnumSerializer<>(SomeEnum.class).snapshotConfiguration();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static EnumSerializerSnapshot<SomeEnum> roundtripSnapshot(
+ EnumSerializerSnapshot<SomeEnum> snapshot, ClassLoader classLoader) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return (EnumSerializerSnapshot<SomeEnum>)
+ TypeSerializerSnapshot.<SomeEnum>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 00000000000..6f5fedb5bae
--- /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<String, Class<?>> 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<SomePojo> read =
+ roundtripSnapshot(writeSnapshot(), getClass().getClassLoader());
+
+ assertThat(read.isPojoClassAvailable()).isTrue();
+ assertThat(read.getPojoClassName()).isEqualTo(SomePojo.class.getName());
+ assertFieldSnapshots(read);
+ }
+
+ @Test
+ void testReadWithClassAbsent() throws IOException {
+ PojoSerializerSnapshot<SomePojo> 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<SomePojo> 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<SomePojo> snapshot) {
+ List<SimpleEntry<String, TypeSerializerSnapshot<?>>> entries =
+ snapshot.getFieldSnapshotEntries();
+
+ assertThat(entries).hasSize(EXPECTED_FIELD_SNAPSHOTS.size());
+ for (SimpleEntry<String, TypeSerializerSnapshot<?>> 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<SomePojo> writeSnapshot() {
+ return (PojoSerializerSnapshot<SomePojo>)
+ TypeExtractor.createTypeInfo(SomePojo.class)
+ .createSerializer(new SerializerConfigImpl())
+ .snapshotConfiguration();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PojoSerializerSnapshot<SomePojo> roundtripSnapshot(
+ PojoSerializerSnapshot<SomePojo> snapshot, ClassLoader classLoader) throws IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return (PojoSerializerSnapshot<SomePojo>)
+ TypeSerializerSnapshot.<SomePojo>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 92580e71b29..4a097ed7e5d 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.reflect.ReflectData;
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 @@ import static org.apache.flink.util.StringUtils.writeString;
* @param <T> The data type that the originating serializer of this configuration serializes.
*/
public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
- private Class<T> runtimeType;
+ @Nullable private Class<T> runtimeType;
+ private String runtimeTypeName;
private Schema schema;
private Schema runtimeSchema;
@@ -61,6 +63,7 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
AvroSerializerSnapshot(Schema schema, Class<T> runtimeType) {
this.schema = schema;
this.runtimeType = runtimeType;
+ this.runtimeTypeName = runtimeType.getName();
}
@Override
@@ -106,7 +109,11 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
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 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
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 class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
@Override
public TypeSerializer<T> 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 class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
}
}
+ /** Returns the Avro writer schema stored in this snapshot. */
+ public Schema getSchema() {
+ return schema;
+ }
+
// ------------------------------------------------------------------------------------------------------------
// Helpers
// ------------------------------------------------------------------------------------------------------------
@@ -225,32 +244,16 @@ public class AvroSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
return d.getSchema(runtimeType);
}
- @SuppressWarnings("unchecked")
- @Nonnull
- private static <T> Class<T> findClassOrThrow(
- ClassLoader userCodeClassLoader, String className) {
- try {
- Class<?> runtimeTarget = Class.forName(className, false, userCodeClassLoader);
- return (Class<T>) 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 <T> Class<T> tryFindClass(ClassLoader userCodeClassLoader, String className) {
+ return CustomRestoreSerializerFactory.resolveOrNull(className, userCodeClassLoader);
}
@SuppressWarnings("unchecked")
- @Nonnull
private static <T> Class<T> findClassOrFallbackToGeneric(
ClassLoader userCodeClassLoader, String className) {
try {
- Class<?> runtimeTarget = Class.forName(className, false, userCodeClassLoader);
- return (Class<T>) runtimeTarget;
+ return (Class<T>) Class.forName(className, false, userCodeClassLoader);
} catch (ClassNotFoundException e) {
return (Class<T>) 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 c0838e9b842..ba7b0231d80 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.isC
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 @@ class AvroSerializerSnapshotTest {
}
}
+ /**
+ * 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<Address> restored = new AvroSerializerSnapshot<>();
+ ClassLoader withoutAddress = classLoaderHiding(Address.class.getSimpleName());
+ restored.readSnapshot(1, new DataInputDeserializer(out.getCopyOfBuffer()), withoutAddress);
+
+ @SuppressWarnings("unchecked")
+ AvroSerializer<Address> serializer = (AvroSerializer<Address>) 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<Address>()
+ .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<Address> 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 @@ class AvroSerializerSnapshotTest {
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 eb027221a0c..d8cc8d1654d 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.
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-type-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<!-- test dependencies -->
<dependency>
@@ -89,7 +96,7 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId>
<version>${project.version}</version>
- <scope>test</scope>
+ <optional>true</optional>
</dependency>
<dependency>
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 00000000000..5105b1d65a0
--- /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.
+ *
+ * <p>{@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<String> {
+
+ 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<String> 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<String> 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 00000000000..2d726f3de26
--- /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}.
+ *
+ * <p>Conversion rules:
+ *
+ * <ul>
+ * <li>{@link String} → {@link StringData}
+ * <li>{@link BigDecimal} → {@link DecimalData} (precision/scale from {@link DecimalType})
+ * <li>{@link ByteBuffer} or {@code byte[]} → {@link DecimalData} (unscaled bytes)
+ * <li>{@link ByteBuffer} → {@code byte[]} for BINARY/VARBINARY
+ * <li>{@link java.sql.Date}, {@link LocalDate} → {@code int} (days since epoch)
+ * <li>{@link Timestamp}, {@link Instant}, {@link LocalDateTime} → {@link TimestampData}
+ * <li>{@link List}, arrays, {@link Iterable} → {@link GenericArrayData} (elements recursively
+ * converted)
+ * <li>{@link Map} or {@link Iterable} of {@link Map.Entry} → {@link GenericMapData} (keys/values
+ * recursively converted)
+ * <li>{@link Row} → {@link GenericRowData} (fields recursively converted)
+ * <li>{@link RowData} subtypes → passed through unchanged
+ * <li>Primitives (boxed) → passed through unchanged
+ * </ul>
+ */
+@Internal
+public final class InternalTypeConverter {
+
+ private InternalTypeConverter() {}
+
+ /**
+ * Converts {@code value} to the Flink table internal representation dictated by {@code type}.
+ *
+ * @param value the raw Java object; may be null
+ * @param type the target logical type; used to drive nested conversions
+ * @return the converted value, or null if value is null
+ */
+ @Nullable
+ public static Object toInternal(@Nullable Object value, LogicalType type) {
+ if (value == null) {
+ return null;
+ }
+
+ switch (type.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ if (value instanceof StringData) {
+ return value;
+ }
+ return StringData.fromString(value.toString());
+
+ case BOOLEAN:
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ case FLOAT:
+ case DOUBLE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case INTERVAL_YEAR_MONTH:
+ case INTERVAL_DAY_TIME:
+ return value;
+
+ case DECIMAL:
+ if (value instanceof DecimalData) {
+ return value;
+ }
+ if (value instanceof BigDecimal) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromBigDecimal(
+ (BigDecimal) value, dt.getPrecision(), dt.getScale());
+ }
+ if (value instanceof ByteBuffer) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromUnscaledBytes(
+ toByteArray((ByteBuffer) value), dt.getPrecision(), dt.getScale());
+ }
+ if (value instanceof byte[]) {
+ DecimalType dt = (DecimalType) type;
+ return DecimalData.fromUnscaledBytes(
+ (byte[]) value, dt.getPrecision(), dt.getScale());
+ }
+ return value;
+
+ case DATE:
+ if (value instanceof Integer) {
+ return value;
+ }
+ if (value instanceof java.sql.Date) {
+ return (int) ((java.sql.Date) value).toLocalDate().toEpochDay();
+ }
+ if (value instanceof LocalDate) {
+ return (int) ((LocalDate) value).toEpochDay();
+ }
+ return value;
+
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ if (value instanceof TimestampData) {
+ return value;
+ }
+ if (value instanceof Timestamp) {
+ return TimestampData.fromTimestamp((Timestamp) value);
+ }
+ if (value instanceof Instant) {
+ return TimestampData.fromInstant((Instant) value);
+ }
+ if (value instanceof LocalDateTime) {
+ return TimestampData.fromLocalDateTime((LocalDateTime) value);
+ }
+ return value;
+
+ case BINARY:
+ case VARBINARY:
+ if (value instanceof ByteBuffer) {
+ return toByteArray((ByteBuffer) value);
+ }
+ return value;
+
+ case NULL:
+ return null;
+
+ case ROW:
+ case STRUCTURED_TYPE:
+ if (value instanceof GenericRowData) {
+ return value;
+ }
+ if (value instanceof Row) {
+ return rowToGenericRowData((Row) value, (RowType) type);
+ }
+ return value;
+
+ case ARRAY:
+ if (value instanceof GenericArrayData) {
+ return value;
+ }
+ ArrayType at = (ArrayType) type;
+ if (value instanceof Object[]) {
+ return objectArrayToArrayData((Object[]) value, at.getElementType());
+ }
+ if (value instanceof Iterable) {
+ return iterableToArrayData((Iterable<?>) value, at.getElementType());
+ }
+ return value;
+
+ case MAP:
+ if (value instanceof GenericMapData) {
+ return value;
+ }
+ MapType mt = (MapType) type;
+ if (value instanceof Map) {
+ return mapToMapData((Map<?, ?>) value, mt.getKeyType(), mt.getValueType());
+ }
+ if (value instanceof Iterable) {
+ return mapEntryIterableToMapData(
+ (Iterable<?>) value, mt.getKeyType(), mt.getValueType());
+ }
+ return value;
+
+ case MULTISET:
+ // MultisetType is not a MapType: it has only an element type, represented
+ // internally as Map<element, Integer> (element -> multiplicity).
+ if (value instanceof GenericMapData) {
+ return value;
+ }
+ LogicalType elementType = ((MultisetType) type).getElementType();
+ if (value instanceof Map) {
+ return mapToMapData((Map<?, ?>) value, elementType, new IntType());
+ }
+ if (value instanceof Iterable) {
+ return mapEntryIterableToMapData(
+ (Iterable<?>) value, elementType, new IntType());
+ }
+ return value;
+
+ default:
+ throw new UnsupportedOperationException(
+ "Cannot convert value of type '"
+ + value.getClass().getName()
+ + "' to internal representation for LogicalTypeRoot "
+ + type.getTypeRoot()
+ + ".");
+ }
+ }
+
+ private static byte[] toByteArray(ByteBuffer bb) {
+ byte[] bytes = new byte[bb.remaining()];
+ bb.get(bytes);
+ return bytes;
+ }
+
+ private static GenericRowData rowToGenericRowData(Row row, RowType rowType) {
+ List<RowType.RowField> fields = rowType.getFields();
+ GenericRowData out = new GenericRowData(row.getArity());
+ out.setRowKind(row.getKind());
+ for (int i = 0; i < row.getArity(); i++) {
+ LogicalType fieldType = i < fields.size() ? fields.get(i).getType() : null;
+ Object rawField = row.getField(i);
+ out.setField(i, fieldType != null ? toInternal(rawField, fieldType) : rawField);
+ }
+ return out;
+ }
+
+ private static GenericArrayData objectArrayToArrayData(Object[] src, LogicalType elementType) {
+ Object[] arr = new Object[src.length];
+ for (int i = 0; i < src.length; i++) {
+ arr[i] = toInternal(src[i], elementType);
+ }
+ return new GenericArrayData(arr);
+ }
+
+ private static GenericArrayData iterableToArrayData(
+ Iterable<?> iterable, LogicalType elementType) {
+ return new GenericArrayData(
+ StreamSupport.stream(iterable.spliterator(), false)
+ .map(v -> toInternal(v, elementType))
+ .toArray());
+ }
+
+ private static GenericMapData mapToMapData(
+ Map<?, ?> map, LogicalType keyType, LogicalType valueType) {
+ LinkedHashMap<Object, Object> converted = new LinkedHashMap<>(map.size());
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ converted.put(
+ toInternal(entry.getKey(), keyType), toInternal(entry.getValue(), valueType));
+ }
+ return new GenericMapData(converted);
+ }
+
+ private static GenericMapData mapEntryIterableToMapData(
+ Iterable<?> iterable, LogicalType keyType, LogicalType valueType) {
+ LinkedHashMap<Object, Object> converted = new LinkedHashMap<>();
+ for (Object element : iterable) {
+ if (!(element instanceof Map.Entry)) {
+ throw new UnsupportedOperationException(
+ "Map conversion supports only Iterable<Map.Entry> but received: "
+ + iterable.getClass().getName());
+ }
+ Map.Entry<?, ?> entry = (Map.Entry<?, ?>) element;
+ converted.put(
+ toInternal(entry.getKey(), keyType), toInternal(entry.getValue(), valueType));
+ }
+ return new GenericMapData(converted);
+ }
+}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoDeserializerCompatibilitySnapshot.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoDeserializerCompatibilitySnapshot.java
new file mode 100644
index 00000000000..eba41a1797c
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoDeserializerCompatibilitySnapshot.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.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.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import javax.annotation.Nullable;
+
+/**
+ * A {@link TypeSerializerSnapshot} for deserializers that can read POJO binary data without the
+ * user POJO class being on the classpath, such as {@link PojoToRowDataDeserializer}. It declares
+ * itself {@link TypeSerializerSchemaCompatibility#compatibleAsIs() compatible as-is} with any
+ * stored {@link PojoSerializerSnapshot}.
+ *
+ * <p>The snapshot only ever exists in memory, wrapping the live deserializer it was created from:
+ * composite compatibility checks (e.g. {@code
+ * CompositeTypeSerializerSnapshot#resolveOuterSchemaCompatibility}) restore the "new" side of a
+ * composite serializer even when nested-level compatibility already short-circuited to {@code
+ * compatibleAsIs()}, so {@link #restoreSerializer()} hands back the wrapped instance rather than
+ * reconstructing one from persisted bytes.
+ */
+@Internal
+public final class PojoDeserializerCompatibilitySnapshot<T> implements TypeSerializerSnapshot<T> {
+
+ @Nullable private final TypeSerializer<T> restoredSerializer;
+
+ /** Constructor for reading the snapshot; see {@link #restoreSerializer()}. */
+ public PojoDeserializerCompatibilitySnapshot() {
+ this(null);
+ }
+
+ public PojoDeserializerCompatibilitySnapshot(TypeSerializer<T> restoredSerializer) {
+ this.restoredSerializer = restoredSerializer;
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return 1;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof PojoSerializerSnapshot
+ || oldSerializerSnapshot instanceof PojoDeserializerCompatibilitySnapshot) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ }
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ @Override
+ public TypeSerializer<T> restoreSerializer() {
+ if (restoredSerializer == null) {
+ throw new UnsupportedOperationException(
+ "PojoDeserializerCompatibilitySnapshot cannot reconstruct the deserializer on "
+ + "its own. Use PojoSerializerSnapshot.restoreSerializer() or "
+ + "PojoToRowDataDeserializer.create().");
+ }
+ return restoredSerializer;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) {}
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) {}
+}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java
new file mode 100644
index 00000000000..390f79bcb5b
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializer.java
@@ -0,0 +1,314 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.api.input.deserializer;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * A {@link TypeSerializer} that reads the POJO binary format written by {@link
+ * org.apache.flink.api.java.typeutils.runtime.PojoSerializer} and produces {@link GenericRowData}.
+ *
+ * <p>This deserializer does <em>not</em> require the user POJO class to be on the classpath. It
+ * mirrors the exact binary protocol of {@code PojoSerializer}:
+ *
+ * <pre>{@code
+ * 1 byte: flags (bitmask)
+ * 0x01 IS_NULL → value is null, return null
+ * 0x02 NO_SUBCLASS → exact POJO class: read numFields × (isNull boolean + field bytes)
+ * 0x08 IS_TAGGED_SUBCLASS → 1 byte subclass tag; delegate to registered subclass deserializer
+ * 0x04 IS_SUBCLASS → UTF class name (must be read); Kryo not supported → throws IOException
+ * }</pre>
+ *
+ * <p>Use {@link #create(PojoSerializerSnapshot)} to build an instance from a savepoint snapshot.
+ */
+@Internal
+public final class PojoToRowDataDeserializer extends TypeSerializer<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(PojoToRowDataDeserializer.class);
+
+ private final int numFields;
+ private final TypeSerializer<?>[] fieldDeserializers;
+ private final LogicalType[] fieldTypes;
+ private final String[] fieldNames;
+ private final List<PojoToRowDataDeserializer> registeredSubclassDeserializers;
+
+ /**
+ * Builds a {@link PojoToRowDataDeserializer} from a {@link PojoSerializerSnapshot}.
+ *
+ * <p>For each field:
+ *
+ * <ul>
+ * <li>If the field snapshot is itself a {@link PojoSerializerSnapshot}, this method recurses
+ * to build a nested {@link PojoToRowDataDeserializer}.
+ * <li>For all other field types, the field's original serializer is restored via {@link
+ * TypeSerializerSnapshot#restoreSerializer()}.
+ * </ul>
+ *
+ * <p>Registered subclasses are handled by building a deserializer for each registered subclass
+ * snapshot in order (matching the tag index used in the binary format).
+ *
+ * @throws IllegalStateException if a required field serializer snapshot is absent
+ */
+ public static PojoToRowDataDeserializer create(PojoSerializerSnapshot<?> snapshot) {
+ List<AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>> fieldEntries =
+ snapshot.getFieldSnapshotEntries();
+
+ List<TypeSerializer<?>> fieldDeserializerList = new ArrayList<>(fieldEntries.size());
+ List<LogicalType> fieldTypeList = new ArrayList<>(fieldEntries.size());
+ List<String> fieldNameList = new ArrayList<>(fieldEntries.size());
+
+ for (AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>> entry : fieldEntries) {
+ String fieldName = entry.getKey();
+ TypeSerializerSnapshot<?> fieldSnapshot = entry.getValue();
+
+ if (fieldSnapshot == null) {
+ throw new IllegalStateException(
+ "Cannot build deserializer for field '"
+ + fieldName
+ + "': its serializer snapshot was not readable from the savepoint. "
+ + "This field cannot be deserialized without the original snapshot.");
+ }
+
+ TypeSerializer<?> fieldDeserializer;
+ if (fieldSnapshot instanceof PojoSerializerSnapshot) {
+ fieldDeserializer = create((PojoSerializerSnapshot<?>) fieldSnapshot);
+ } else {
+ fieldDeserializer = fieldSnapshot.restoreSerializer();
+ }
+
+ fieldDeserializerList.add(fieldDeserializer);
+ fieldTypeList.add(SerializerSnapshotToLogicalTypeConverter.convert(fieldSnapshot));
+ fieldNameList.add(fieldName);
+ }
+
+ List<TypeSerializerSnapshot<?>> subSnapshots =
+ snapshot.getRegisteredSubclassSnapshotsOrdered();
+ List<PojoToRowDataDeserializer> subDeserializers = new ArrayList<>(subSnapshots.size());
+ for (TypeSerializerSnapshot<?> subSnap : subSnapshots) {
+ subDeserializers.add(
+ subSnap instanceof PojoSerializerSnapshot
+ ? create((PojoSerializerSnapshot<?>) subSnap)
+ : null);
+ }
+
+ return new PojoToRowDataDeserializer(
+ fieldDeserializerList.toArray(new TypeSerializer[0]),
+ fieldTypeList.toArray(new LogicalType[0]),
+ fieldNameList.toArray(new String[0]),
+ subDeserializers);
+ }
+
+ PojoToRowDataDeserializer(
+ TypeSerializer<?>[] fieldDeserializers,
+ LogicalType[] fieldTypes,
+ String[] fieldNames,
+ List<PojoToRowDataDeserializer> registeredSubclassDeserializers) {
+ this.numFields = fieldDeserializers.length;
+ this.fieldDeserializers = fieldDeserializers;
+ this.fieldTypes = fieldTypes;
+ this.fieldNames = fieldNames;
+ this.registeredSubclassDeserializers = registeredSubclassDeserializers;
+ }
+
+ @Override
+ public RowData deserialize(DataInputView source) throws IOException {
+ int flags = source.readByte() & 0xFF;
+
+ if ((flags & PojoSerializer.IS_NULL) != 0) {
+ return null;
+ }
+
+ if ((flags & PojoSerializer.NO_SUBCLASS) != 0) {
+ return readFields(source);
+ }
+
+ if ((flags & PojoSerializer.IS_TAGGED_SUBCLASS) != 0) {
+ int tag = source.readByte() & 0xFF;
+ if (tag < registeredSubclassDeserializers.size()) {
+ PojoToRowDataDeserializer subDeserializer =
+ registeredSubclassDeserializers.get(tag);
+ if (subDeserializer == null) {
+ // Either the subclass's own snapshot was unreadable, or the subclass is not
+ // itself a POJO (e.g. it falls back to Kryo) — either way we have no way to
+ // decode its bytes, and, like the IS_SUBCLASS/Kryo case below, its length is
+ // unknown so the bytes cannot even be skipped.
+ throw new IOException(
+ "Cannot deserialize registered POJO subclass at tag "
+ + tag
+ + ": its serializer snapshot is missing or is not a POJO "
+ + "serializer (e.g. it uses Kryo), which requires the class on "
+ + "the classpath.");
+ }
+ return subDeserializer.deserialize(source);
+ }
+ throw new IOException(
+ "Unknown registered subclass tag "
+ + tag
+ + " (have "
+ + registeredSubclassDeserializers.size()
+ + " registered). The savepoint may have been written with more subclasses registered.");
+ }
+
+ if ((flags & PojoSerializer.IS_SUBCLASS) != 0) {
+ String className = source.readUTF();
+ throw new IOException(
+ "Cannot deserialize POJO subclass '"
+ + className
+ + "': the subclass uses Kryo serialization, which requires the class on the"
+ + " classpath. Kryo-encoded bytes have unknown length and cannot be skipped."
+ + " Register the subclass or add the JAR to the classpath.");
+ }
+
+ throw new IOException("Unrecognised POJO flags byte: 0x" + Integer.toHexString(flags));
+ }
+
+ @Override
+ public RowData deserialize(RowData reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ private GenericRowData readFields(DataInputView source) throws IOException {
+ GenericRowData row = new GenericRowData(numFields);
+ for (int i = 0; i < numFields; i++) {
+ boolean isNull = source.readBoolean();
+ if (isNull) {
+ row.setField(i, null);
+ continue;
+ }
+ // Unlike the conversion step below, a deserialize() failure here means the stream
+ // position for this and every subsequent field/row is now unknown: continuing to read
+ // would silently cascade garbage into later rows. Fail loudly instead, mirroring
+ // PojoSerializer.deserialize(), which never catches per-field failures either.
+ Object raw = fieldDeserializers[i].deserialize(source);
+ try {
+ row.setField(i, InternalTypeConverter.toInternal(raw, fieldTypes[i]));
+ } catch (Exception e) {
+ // Bytes were already consumed correctly, so the stream is still aligned for
+ // subsequent fields/rows; only this field's value could not be mapped to its
+ // table-internal representation. Safe to null just this field and continue.
+ LOG.warn(
+ "Failed to convert field '{}' (index {}) value: {}. Setting field to null.",
+ fieldNames[i],
+ i,
+ e.getMessage());
+ row.setField(i, null);
+ }
+ }
+ return row;
+ }
+
+ // -------------------------------------------------------------------------
+ // TypeSerializer boilerplate — copy/snapshot operations not needed for
+ // schema-extraction use cases but required by the interface.
+ // -------------------------------------------------------------------------
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<RowData> duplicate() {
+ return this;
+ }
+
+ @Override
+ public RowData createInstance() {
+ return new GenericRowData(numFields);
+ }
+
+ @Override
+ public RowData copy(RowData from) {
+ return from;
+ }
+
+ @Override
+ public RowData copy(RowData from, RowData reuse) {
+ return from;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(RowData record, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "PojoToRowDataDeserializer is read-only; serialization is not supported.");
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException(
+ "PojoToRowDataDeserializer is read-only; copy is not supported.");
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof PojoToRowDataDeserializer)) {
+ return false;
+ }
+ PojoToRowDataDeserializer other = (PojoToRowDataDeserializer) obj;
+ return numFields == other.numFields
+ && Arrays.equals(fieldDeserializers, other.fieldDeserializers)
+ && Arrays.equals(fieldTypes, other.fieldTypes)
+ && Arrays.equals(fieldNames, other.fieldNames)
+ && registeredSubclassDeserializers.equals(other.registeredSubclassDeserializers);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = numFields;
+ result = 31 * result + Arrays.hashCode(fieldDeserializers);
+ result = 31 * result + Arrays.hashCode(fieldTypes);
+ result = 31 * result + Arrays.hashCode(fieldNames);
+ result = 31 * result + registeredSubclassDeserializers.hashCode();
+ return result;
+ }
+
+ @Override
+ public TypeSerializerSnapshot<RowData> snapshotConfiguration() {
+ return new PojoDeserializerCompatibilitySnapshot<>(this);
+ }
+}
diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java
index fa8b73340ea..5936b30bb5d 100644
--- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java
+++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/KeyedStateReadingITCase.java
@@ -18,14 +18,23 @@
package org.apache.flink.state.api;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.checkpoint.OperatorState;
import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
+import org.apache.flink.state.api.input.deserializer.PojoToRowDataDeserializer;
import org.apache.flink.state.api.runtime.SavepointLoader;
import org.apache.flink.state.api.schema.KeyedStateSchemaInfo;
+import org.apache.flink.state.api.schema.StateSchemaExtractor;
+import org.apache.flink.state.api.schema.StateSchemaInfo;
import org.apache.flink.state.api.utils.SavepointTestBase;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
@@ -33,13 +42,16 @@ import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.Objects;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Integration tests that write real keyed state through a MiniCluster job, take a savepoint at
@@ -52,10 +64,37 @@ public abstract class KeyedStateReadingITCase extends SavepointTestBase {
protected abstract Configuration getConfiguration();
+ /** Deliberately simple so no Kryo or special serializers are needed. */
+ public static class PersonPojo {
+ public String name;
+ public int age;
+ public long score;
+
+ public PersonPojo() {}
+
+ public PersonPojo(String name, int age, long score) {
+ this.name = name;
+ this.age = age;
+ this.score = score;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof PersonPojo)) {
+ return false;
+ }
+ PersonPojo other = (PersonPojo) o;
+ return Objects.equals(name, other.name) && age == other.age && score == other.score;
+ }
+ }
+
// -------------------------------------------------------------------------
// Schema extraction: RowData-typed internal SQL operator state
// -------------------------------------------------------------------------
+ private static final ValueStateDescriptor<PersonPojo> PERSON_STATE_DESC =
+ new ValueStateDescriptor<>("person", PersonPojo.class);
+
@Test
public void testGroupAggAccStateSchemaExtraction() throws Exception {
StreamExecutionEnvironment env =
@@ -108,4 +147,137 @@ public abstract class KeyedStateReadingITCase extends SavepointTestBase {
assertEquals(LogicalTypeRoot.BIGINT, rowType.getFields().get(0).getType().getTypeRoot());
assertEquals(LogicalTypeRoot.BIGINT, rowType.getFields().get(1).getType().getTypeRoot());
}
+
+ // -------------------------------------------------------------------------
+ // Schema extraction: POJO value state
+ // -------------------------------------------------------------------------
+
+ private static final String POJO_UID = "pojo-state-operator";
+
+ @Test
+ public void testSchemaExtractionFromPojoState() throws Exception {
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(getConfiguration());
+ env.setParallelism(1);
+
+ PersonPojo[] data = {
+ new PersonPojo("Alice", 30, 100L),
+ new PersonPojo("Bob", 25, 200L),
+ new PersonPojo("Carol", 35, 300L)
+ };
+ env.addSource(createSource(data))
+ .returns(PersonPojo.class)
+ .keyBy(p -> p.name)
+ .process(new PersonStateWriter())
+ .uid(POJO_UID)
+ .sinkTo(new DiscardingSink<>());
+
+ String savepointPath = takeSavepoint(env);
+ CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(savepointPath);
+
+ OperatorIdentifier opId = OperatorIdentifier.forUid(POJO_UID);
+
+ // Discover states via StateTableUtils
+ List<String> stateNames = StateTableUtils.getKeyedStates(metadata, opId);
+ assertTrue(stateNames.contains("person"), "Expected 'person' state");
+
+ // Extract schema via StateTableUtils — all states in one call
+ KeyedStateSchemaInfo schemaInfo = StateTableUtils.getKeyedStateSchema(metadata, opId);
+ KeyedStateSchemaInfo.StateEntryInfo personEntry = schemaInfo.stateSchemas.get("person");
+ assertNotNull(personEntry, "'person' state not found in schema");
+
+ // PersonPojo has 3 fields → the logicalType should be a RowType with 3 fields
+ assertEquals(LogicalTypeRoot.ROW, personEntry.logicalType.getTypeRoot());
+ RowType rowType = (RowType) personEntry.logicalType;
+ assertEquals(3, rowType.getFieldCount());
+ assertHasField(rowType, "name", LogicalTypeRoot.VARCHAR);
+ assertHasField(rowType, "age", LogicalTypeRoot.INTEGER);
+ assertHasField(rowType, "score", LogicalTypeRoot.BIGINT);
+ }
+
+ @Test
+ public void testDeserializerBuiltFromPojoSnapshot() throws Exception {
+ StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(getConfiguration());
+ env.setParallelism(1);
+
+ PersonPojo[] data = {new PersonPojo("Alice", 30, 100L)};
+ env.addSource(createSource(data))
+ .returns(PersonPojo.class)
+ .keyBy(p -> p.name)
+ .process(new PersonStateWriter())
+ .uid(POJO_UID)
+ .sinkTo(new DiscardingSink<>());
+
+ String savepointPath = takeSavepoint(env);
+ CheckpointMetadata metadata = SavepointLoader.loadSavepointMetadata(savepointPath);
+
+ OperatorIdentifier opId = OperatorIdentifier.forUid(POJO_UID);
+
+ KeyedStateSchemaInfo schemaInfo = StateTableUtils.getKeyedStateSchema(metadata, opId);
+ KeyedStateSchemaInfo.StateEntryInfo personEntry = schemaInfo.stateSchemas.get("person");
+ assertNotNull(personEntry);
+
+ // Building the PojoToRowDataDeserializer directly from the snapshot (lower-level API)
+ List<StateSchemaInfo> rawSchemas =
+ StateSchemaExtractor.extractSchema(findOperatorState(metadata, opId));
+ StateSchemaInfo personRaw =
+ rawSchemas.stream()
+ .filter(s -> "person".equals(s.stateName))
+ .findFirst()
+ .orElse(null);
+ assertNotNull(personRaw);
+
+ var deser =
+ PojoToRowDataDeserializer.create(
+ (PojoSerializerSnapshot<?>) personRaw.valueSnapshot);
+ assertNotNull(deser);
+ assertTrue(
+ deser instanceof PojoToRowDataDeserializer,
+ "Expected PojoToRowDataDeserializer, got: " + deser.getClass().getSimpleName());
+ }
+
+ // -------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------
+
+ private static OperatorState findOperatorState(
+ CheckpointMetadata metadata, OperatorIdentifier opId) {
+ for (OperatorState op : metadata.getOperatorStates()) {
+ if (op.getOperatorID().equals(opId.getOperatorId())) {
+ return op;
+ }
+ }
+ throw new IllegalArgumentException("Operator not found: " + opId);
+ }
+
+ private static void assertHasField(RowType row, String name, LogicalTypeRoot expectedRoot) {
+ RowType.RowField field =
+ row.getFields().stream()
+ .filter(f -> f.getName().equals(name))
+ .findFirst()
+ .orElse(null);
+ assertNotNull(field, "Field '" + name + "' not found in row type");
+ assertEquals(
+ expectedRoot, field.getType().getTypeRoot(), "Wrong type for field '" + name + "'");
+ }
+
+ // -------------------------------------------------------------------------
+ // Operators
+ // -------------------------------------------------------------------------
+
+ private static class PersonStateWriter extends KeyedProcessFunction<String, PersonPojo, Void> {
+ private transient ValueState<PersonPojo> state;
+
+ @Override
+ public void open(OpenContext ctx) throws Exception {
+ state = getRuntimeContext().getState(PERSON_STATE_DESC);
+ }
+
+ @Override
+ public void processElement(PersonPojo value, Context ctx, Collector<Void> out)
+ throws Exception {
+ state.update(value);
+ }
+ }
}
diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverterTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverterTest.java
new file mode 100644
index 00000000000..6efe86b3f86
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/InternalTypeConverterTest.java
@@ -0,0 +1,301 @@
+/*
+ * 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.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.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DayTimeIntervalType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+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.NullType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimeType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.YearMonthIntervalType;
+import org.apache.flink.table.types.logical.ZonedTimestampType;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.Test;
+
+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.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/** Unit tests for {@link InternalTypeConverter}. */
+public class InternalTypeConverterTest {
+
+ @Test
+ public void testNullReturnsNull() {
+ assertNull(InternalTypeConverter.toInternal(null, new IntType()));
+ assertNull(InternalTypeConverter.toInternal(null, new VarCharType()));
+ assertNull(InternalTypeConverter.toInternal("anything", new NullType()));
+ }
+
+ @Test
+ public void testVarChar() {
+ // String → StringData
+ assertEquals(
+ StringData.fromString("hello"),
+ InternalTypeConverter.toInternal("hello", new VarCharType()));
+ // StringData → pass-through
+ StringData sd = StringData.fromString("world");
+ assertSame(sd, InternalTypeConverter.toInternal(sd, new VarCharType()));
+ // Other type → toString()
+ assertEquals(
+ StringData.fromString("42"),
+ InternalTypeConverter.toInternal(42, new VarCharType()));
+ }
+
+ @Test
+ public void testPrimitivePassThroughs() {
+ // All of these are returned unchanged.
+ assertSame(Boolean.TRUE, InternalTypeConverter.toInternal(true, new BooleanType()));
+ Byte b = (byte) 7;
+ assertSame(b, InternalTypeConverter.toInternal(b, new TinyIntType()));
+ Short s = (short) 100;
+ assertSame(s, InternalTypeConverter.toInternal(s, new SmallIntType()));
+ Integer i = 42;
+ assertSame(i, InternalTypeConverter.toInternal(i, new IntType()));
+ Long l = 123L;
+ assertSame(l, InternalTypeConverter.toInternal(l, new BigIntType()));
+ Float f = 1.5f;
+ assertSame(f, InternalTypeConverter.toInternal(f, new FloatType()));
+ Double d = 3.14;
+ assertSame(d, InternalTypeConverter.toInternal(d, new DoubleType()));
+ Integer timeMillis = 3_600_000;
+ assertSame(timeMillis, InternalTypeConverter.toInternal(timeMillis, new TimeType()));
+ Long months = 13L;
+ assertSame(
+ months,
+ InternalTypeConverter.toInternal(
+ months,
+ new YearMonthIntervalType(
+ YearMonthIntervalType.YearMonthResolution.YEAR_TO_MONTH)));
+ Long dayMillis = 86_400_000L;
+ assertSame(
+ dayMillis,
+ InternalTypeConverter.toInternal(
+ dayMillis,
+ new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY)));
+ }
+
+ @Test
+ public void testDecimal() {
+ DecimalType type = new DecimalType(10, 2);
+ // BigDecimal → DecimalData
+ BigDecimal bd = new BigDecimal("12.34");
+ assertEquals(
+ DecimalData.fromBigDecimal(bd, 10, 2), InternalTypeConverter.toInternal(bd, type));
+ // byte[] → DecimalData (unscaled bytes)
+ byte[] unscaledBytes = BigDecimal.valueOf(1234).unscaledValue().toByteArray();
+ assertEquals(
+ DecimalData.fromUnscaledBytes(unscaledBytes, 10, 2),
+ InternalTypeConverter.toInternal(unscaledBytes, type));
+ // ByteBuffer → DecimalData (unscaled bytes)
+ assertEquals(
+ DecimalData.fromUnscaledBytes(unscaledBytes, 10, 2),
+ InternalTypeConverter.toInternal(ByteBuffer.wrap(unscaledBytes), type));
+ // DecimalData → pass-through
+ DecimalData dd = DecimalData.fromBigDecimal(new BigDecimal("9.99"), 10, 2);
+ assertSame(dd, InternalTypeConverter.toInternal(dd, type));
+ }
+
+ @Test
+ public void testDate() {
+ // Integer (epoch day) → pass-through
+ Integer epochDay = 19_000;
+ assertSame(epochDay, InternalTypeConverter.toInternal(epochDay, new DateType()));
+ // LocalDate → epoch day int
+ LocalDate ld = LocalDate.of(2022, 6, 15);
+ assertEquals((int) ld.toEpochDay(), InternalTypeConverter.toInternal(ld, new DateType()));
+ // java.sql.Date → epoch day int
+ java.sql.Date sqlDate = java.sql.Date.valueOf("2022-06-15");
+ assertEquals(
+ (int) sqlDate.toLocalDate().toEpochDay(),
+ InternalTypeConverter.toInternal(sqlDate, new DateType()));
+ }
+
+ @Test
+ public void testTimestamp() {
+ Timestamp ts = Timestamp.valueOf("2023-01-15 10:30:00");
+ Instant instant = Instant.parse("2023-01-15T10:30:00Z");
+ LocalDateTime ldt = LocalDateTime.of(2023, 1, 15, 10, 30, 0);
+
+ // All three timestamp type roots accept the same source types.
+ for (LogicalType tsType :
+ new LogicalType[] {
+ new TimestampType(), new ZonedTimestampType(), new LocalZonedTimestampType()
+ }) {
+ assertEquals(
+ TimestampData.fromTimestamp(ts), InternalTypeConverter.toInternal(ts, tsType));
+ assertEquals(
+ TimestampData.fromInstant(instant),
+ InternalTypeConverter.toInternal(instant, tsType));
+ assertEquals(
+ TimestampData.fromLocalDateTime(ldt),
+ InternalTypeConverter.toInternal(ldt, tsType));
+ }
+ // TimestampData → pass-through
+ TimestampData td = TimestampData.fromEpochMillis(1000L);
+ assertSame(td, InternalTypeConverter.toInternal(td, new TimestampType()));
+ }
+
+ @Test
+ public void testBinary() {
+ byte[] bytes = {1, 2, 3};
+ // byte[] → pass-through
+ assertSame(bytes, InternalTypeConverter.toInternal(bytes, new VarBinaryType()));
+ // ByteBuffer → extracted byte[]
+ assertArrayEquals(
+ bytes,
+ (byte[])
+ InternalTypeConverter.toInternal(
+ ByteBuffer.wrap(bytes), new VarBinaryType()));
+ }
+
+ @Test
+ public void testRow() {
+ RowType rowType = RowType.of(new VarCharType(), new IntType());
+ // Flink Row → GenericRowData with recursive field conversion
+ Row row = Row.ofKind(RowKind.INSERT, "Alice", 30);
+ GenericRowData result = (GenericRowData) InternalTypeConverter.toInternal(row, rowType);
+ assertEquals(StringData.fromString("Alice"), result.getString(0));
+ assertEquals(30, result.getInt(1));
+ // GenericRowData → pass-through
+ GenericRowData grd = GenericRowData.of(StringData.fromString("x"), 1);
+ assertSame(grd, InternalTypeConverter.toInternal(grd, rowType));
+ }
+
+ @Test
+ public void testArray() {
+ ArrayType intArrayType = new ArrayType(new IntType());
+ ArrayType strArrayType = new ArrayType(new VarCharType());
+
+ // List → GenericArrayData
+ GenericArrayData fromList =
+ (GenericArrayData)
+ InternalTypeConverter.toInternal(Arrays.asList(1, 2, 3), intArrayType);
+ assertEquals(3, fromList.size());
+ assertEquals(1, fromList.getInt(0));
+ assertEquals(3, fromList.getInt(2));
+
+ // Object[] → GenericArrayData with recursive element conversion
+ GenericArrayData fromObjectArray =
+ (GenericArrayData)
+ InternalTypeConverter.toInternal(new Object[] {"a", "b"}, strArrayType);
+ assertEquals(StringData.fromString("a"), fromObjectArray.getString(0));
+ assertEquals(StringData.fromString("b"), fromObjectArray.getString(1));
+
+ // Iterable → GenericArrayData (ListState returns Iterable)
+ GenericArrayData fromIterable =
+ (GenericArrayData)
+ InternalTypeConverter.toInternal(
+ Arrays.asList(10L, 20L), new ArrayType(new BigIntType()));
+ assertEquals(10L, fromIterable.getLong(0));
+ assertEquals(20L, fromIterable.getLong(1));
+
+ // GenericArrayData → pass-through
+ GenericArrayData gad = new GenericArrayData(new Object[] {1, 2});
+ assertSame(gad, InternalTypeConverter.toInternal(gad, intArrayType));
+ }
+
+ @Test
+ public void testMap() {
+ MapType type = new MapType(new VarCharType(), new IntType());
+
+ // Map → GenericMapData with recursive key/value conversion
+ Map<String, Integer> map = new LinkedHashMap<>();
+ map.put("a", 1);
+ map.put("b", 2);
+ GenericMapData fromMap = (GenericMapData) InternalTypeConverter.toInternal(map, type);
+ assertEquals(2, fromMap.size());
+ assertEquals(1, fromMap.get(StringData.fromString("a")));
+ assertEquals(2, fromMap.get(StringData.fromString("b")));
+
+ // Iterable<Map.Entry> → GenericMapData (MapState.entries() returns this)
+ GenericMapData fromEntries =
+ (GenericMapData) InternalTypeConverter.toInternal(map.entrySet(), type);
+ assertEquals(1, fromEntries.get(StringData.fromString("a")));
+ assertEquals(2, fromEntries.get(StringData.fromString("b")));
+
+ // GenericMapData → pass-through
+ Map<Object, Object> inner = new HashMap<>();
+ inner.put(StringData.fromString("k"), 99);
+ GenericMapData gmd = new GenericMapData(inner);
+ assertSame(gmd, InternalTypeConverter.toInternal(gmd, type));
+ }
+
+ @Test
+ public void testMultiset() {
+ // MultisetType is not a MapType: it only carries an element type, and is represented
+ // internally as Map<element, Integer> (element -> multiplicity).
+ MultisetType type = new MultisetType(new VarCharType());
+
+ Map<String, Integer> map = new LinkedHashMap<>();
+ map.put("a", 3);
+ map.put("b", 1);
+ GenericMapData fromMap = (GenericMapData) InternalTypeConverter.toInternal(map, type);
+ assertEquals(2, fromMap.size());
+ assertEquals(3, fromMap.get(StringData.fromString("a")));
+ assertEquals(1, fromMap.get(StringData.fromString("b")));
+
+ // Iterable<Map.Entry> → GenericMapData
+ GenericMapData fromEntries =
+ (GenericMapData) InternalTypeConverter.toInternal(map.entrySet(), type);
+ assertEquals(3, fromEntries.get(StringData.fromString("a")));
+ assertEquals(1, fromEntries.get(StringData.fromString("b")));
+
+ // GenericMapData → pass-through
+ Map<Object, Object> inner = new HashMap<>();
+ inner.put(StringData.fromString("k"), 5);
+ GenericMapData gmd = new GenericMapData(inner);
+ assertSame(gmd, InternalTypeConverter.toInternal(gmd, type));
+ }
+}
diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializerTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializerTest.java
new file mode 100644
index 00000000000..37905ce0220
--- /dev/null
+++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/deserializer/PojoToRowDataDeserializerTest.java
@@ -0,0 +1,239 @@
+/*
+ * 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.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.state.api.schema.SerializerSnapshotToLogicalTypeConverter;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Unit tests for {@link PojoToRowDataDeserializer}.
+ *
+ * <p>Each test serializes a POJO using the real {@link PojoSerializer}, then deserializes with
+ * {@link PojoToRowDataDeserializer} — no POJO class needed on the deserialization side.
+ */
+public class PojoToRowDataDeserializerTest {
+
+ // -------------------------------------------------------------------------
+ // POJO classes
+ // -------------------------------------------------------------------------
+
+ public static class FlatPojo {
+ public String name;
+ public int age;
+ public long score;
+ public boolean active;
+
+ public FlatPojo() {}
+
+ public FlatPojo(String name, int age, long score, boolean active) {
+ this.name = name;
+ this.age = age;
+ this.score = score;
+ this.active = active;
+ }
+ }
+
+ public static class PojoWithNullableField {
+ public String tag; // may be null
+ public int value;
+
+ public PojoWithNullableField() {}
+
+ public PojoWithNullableField(String tag, int value) {
+ this.tag = tag;
+ this.value = value;
+ }
+ }
+
+ public static class NestedPojo {
+ public String label;
+ public FlatPojo inner;
+
+ public NestedPojo() {}
+
+ public NestedPojo(String label, FlatPojo inner) {
+ this.label = label;
+ this.inner = inner;
+ }
+ }
+
+ // -------------------------------------------------------------------------
+ // Tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ public void testDeserializeFlatPojo() throws IOException {
+ FlatPojo original = new FlatPojo("Alice", 30, 12345L, true);
+
+ PojoToRowDataDeserializer deserializer = buildDeserializer(FlatPojo.class);
+ GenericRowData row = (GenericRowData) roundtrip(original, FlatPojo.class, deserializer);
+
+ assertNotNull(row);
+ assertEquals(4, row.getArity());
+ assertEquals(
+ StringData.fromString("Alice"),
+ row.getString(indexOfField(FlatPojo.class, "name")));
+ assertEquals(30, row.getInt(indexOfField(FlatPojo.class, "age")));
+ assertEquals(12345L, row.getLong(indexOfField(FlatPojo.class, "score")));
+ assertTrue(row.getBoolean(indexOfField(FlatPojo.class, "active")));
+ }
+
+ @Test
+ public void testDeserializeWithNullField() throws IOException {
+ PojoWithNullableField original = new PojoWithNullableField(null, 42);
+ PojoToRowDataDeserializer deserializer = buildDeserializer(PojoWithNullableField.class);
+ GenericRowData row =
+ (GenericRowData) roundtrip(original, PojoWithNullableField.class, deserializer);
+
+ assertNotNull(row);
+ assertTrue(row.isNullAt(indexOfField(PojoWithNullableField.class, "tag")));
+ assertEquals(42, row.getInt(indexOfField(PojoWithNullableField.class, "value")));
+ }
+
+ @Test
+ public void testDeserializeNullValue() throws IOException {
+ TypeSerializer<FlatPojo> pojoSer = buildPojoSerializer(FlatPojo.class);
+ DataOutputSerializer out = new DataOutputSerializer(64);
+ pojoSer.serialize(null, out);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ PojoToRowDataDeserializer deserializer = buildDeserializer(FlatPojo.class);
+ RowData result = deserializer.deserialize(in);
+ assertNull(result);
+ }
+
+ @Test
+ public void testDeserializeNestedPojo() throws IOException {
+ NestedPojo original = new NestedPojo("outer", new FlatPojo("Bob", 25, 999L, false));
+ PojoToRowDataDeserializer deserializer = buildDeserializer(NestedPojo.class);
+ GenericRowData row = (GenericRowData) roundtrip(original, NestedPojo.class, deserializer);
+
+ assertNotNull(row);
+ int labelIdx = indexOfField(NestedPojo.class, "label");
+ int innerIdx = indexOfField(NestedPojo.class, "inner");
+ assertEquals(StringData.fromString("outer"), row.getString(labelIdx));
+
+ // Nested POJO should be a GenericRowData
+ RowData innerRow = row.getRow(innerIdx, 4);
+ assertNotNull(innerRow);
+ }
+
+ @Test
+ public void testUnregisteredSubclassThrowsIoException() throws IOException {
+ // Write a value normally using the serializer, then inject fake IS_SUBCLASS bytes.
+ DataOutputSerializer out = new DataOutputSerializer(64);
+ out.writeByte(PojoSerializer.IS_SUBCLASS);
+ out.writeUTF("com.example.UnknownSubclass");
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ PojoToRowDataDeserializer deserializer = buildDeserializer(FlatPojo.class);
+
+ IOException e = assertThrows(IOException.class, () -> deserializer.deserialize(in));
+ assertTrue(e.getMessage().contains("UnknownSubclass"));
+ }
+
+ @Test
+ public void testTaggedSubclassWithUnresolvableDeserializerThrowsIoException()
+ throws IOException {
+ // A registered subclass whose own serializer snapshot could not be turned into a
+ // PojoToRowDataDeserializer (e.g. it is not a POJO, or its snapshot was unreadable) is
+ // represented by a `null` entry in registeredSubclassDeserializers (see
+ // PojoSerializerSnapshot#getRegisteredSubclassSnapshotsOrdered). Deserializing a tagged
+ // subclass that resolves to such an entry must fail with a clear IOException rather than
+ // an NPE.
+ List<PojoToRowDataDeserializer> registeredSubclassDeserializers = new ArrayList<>();
+ registeredSubclassDeserializers.add(null);
+ PojoToRowDataDeserializer deserializer =
+ new PojoToRowDataDeserializer(
+ new TypeSerializer[0],
+ new LogicalType[0],
+ new String[0],
+ registeredSubclassDeserializers);
+
+ DataOutputSerializer out = new DataOutputSerializer(8);
+ out.writeByte(PojoSerializer.IS_TAGGED_SUBCLASS);
+ out.writeByte(0);
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+
+ IOException e = assertThrows(IOException.class, () -> deserializer.deserialize(in));
+ assertTrue(e.getMessage().contains("tag 0"));
+ }
+
+ // -------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------
+
+ @SuppressWarnings("unchecked")
+ private static <T> PojoSerializer<T> buildPojoSerializer(Class<T> clazz) {
+ return (PojoSerializer<T>)
+ TypeExtractor.createTypeInfo(clazz).createSerializer(new SerializerConfigImpl());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> PojoToRowDataDeserializer buildDeserializer(Class<T> clazz) {
+ PojoSerializer<T> ser = buildPojoSerializer(clazz);
+ PojoSerializerSnapshot<T> snapshot =
+ (PojoSerializerSnapshot<T>) ser.snapshotConfiguration();
+ return PojoToRowDataDeserializer.create(snapshot);
+ }
+
+ private static <T> RowData roundtrip(T value, Class<T> clazz, PojoToRowDataDeserializer deser)
+ throws IOException {
+ PojoSerializer<T> ser = buildPojoSerializer(clazz);
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ ser.serialize(value, out);
+
+ DataInputDeserializer in = new DataInputDeserializer(out.getSharedBuffer());
+ return deser.deserialize(in);
+ }
+
+ /** Returns the field index as it appears in the PojoSerializer's field ordering. */
+ private static int indexOfField(Class<?> clazz, String fieldName) {
+ RowType rowType =
+ (RowType)
+ SerializerSnapshotToLogicalTypeConverter.convert(
+ buildPojoSerializer(clazz).snapshotConfiguration());
+ int idx = rowType.getFieldNames().indexOf(fieldName);
+ assertTrue(idx >= 0, "Field '" + fieldName + "' not found");
+ return idx;
+ }
+}
diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
index 718fbc20be9..6e0944975fa 100644
--- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
+++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java
@@ -407,5 +407,22 @@ public class RowDataSerializer extends AbstractRowDataSerializer<RowData> {
return intermediateResult.getFinalResult();
}
+
+ /** Returns the logical types stored in this snapshot. */
+ @Internal
+ public LogicalType[] getTypes() {
+ return types;
+ }
+
+ /**
+ * Returns the field names stored in this snapshot, in the same order as {@link
+ * #getTypes()}, or {@code null} if the originating serializer was built from a bare {@link
+ * LogicalType} array or the snapshot predates field name tracking.
+ */
+ @Internal
+ @Nullable
+ public String[] getFieldNames() {
+ return fieldNames;
+ }
}
}