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 <[email protected]> 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; + } } }
