[FLINK-40175][State Catalog] State Catalog initial catalog functionality and keyed state support

docs/content/docs/libs/state_processor_api.md

@@ -749,17 +749,13 @@ The following predicates on the key column can be pushed down:
 | Option                           | Required | Default | Type                                   | Description                                                                                                                                                                                                                                                                      |
 |----------------------------------|----------|---------|----------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | fields.#.state-name              | optional | (none)  | String                                 | Overrides the state name which must be used for state reading. This can be useful when the state name contains characters which are not compliant with SQL column names.                                                                                                         |
-| fields.#.state-type              | optional | (none)  | Enum Possible values: list, map, value | Defines the state type which must be used for state reading, including value, list and map. When it's not provided then it tries to infer from the SQL type (ARRAY=list, MAP=map, all others=value).                                                                             |
-| fields.#.key-class               | optional | (none)  | String                                 | Defines the format class scheme for decoding map key data (for ex. java.lang.Long). Either key-class or key-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported).        |
-| fields.#.key-type-factory        | optional | (none)  | String                                 | Defines the type information factory for decoding map key data. Either key-class or key-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported).                            |
-| fields.#.value-class             | optional | (none)  | String                                 | Defines the format class scheme for decoding value data (for ex. java.lang.Long). Either value-class or value-info-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported).      |
-| fields.#.value-type-factory      | optional | (none)  | String                                 | Defines the type information factory for decoding value data. Either value-class or value-type-factory can be specified. When none of them are provided then the format class scheme tries to infer from the SQL type (only primitive types supported).                          |
 
 ### Default Data Type Mapping
 
-The state SQL connector infers the data type for primitive types when `fields.#.value-class` and `fields.#.key-class`
-are not defined. The following table shows the `Flink SQL type` -> `Java type` default mapping. If the mapping is not calculated properly
-then it can be overridden with the two mentioned config parameters on a per-column basis.
+The state SQL connector automatically infers each column's data type from the savepoint's serializer
+snapshot metadata (this covers primitive, Avro, Row and POJO types). When no serializer snapshot is
+available for a column, it falls back to inferring a primitive Java type directly from the SQL type,
+using the mapping below.
 
 | Flink SQL type          | Java type                                                               |
 |-------------------------|-------------------------------------------------------------------------|

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;
+    }
+}

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/EnumSerializer.java

@@ -20,12 +20,12 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.util.InstantiationUtil;
 
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -184,6 +184,8 @@ public static final class EnumSerializerSnapshot<T extends Enum<T>>
 
         private T[] enums;
         private Class<T> enumClass;
+        private String enumClassName;
+        private String[] enumNames;
 
         @SuppressWarnings("unused")
         public EnumSerializerSnapshot() {
@@ -213,31 +215,49 @@ public void writeSnapshot(DataOutputView out) throws IOException {
         @Override
         public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
                 throws IOException {
-            enumClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
+            final String className = in.readUTF();
+            enumClass =
+                    CustomRestoreSerializerFactory.resolveOrNull(className, userCodeClassLoader);
+            enumClassName = className;
 
             int numEnumConstants = in.readInt();
-
             @SuppressWarnings("unchecked")
-            T[] previousEnums = (T[]) Array.newInstance(enumClass, numEnumConstants);
+            T[] previousEnums =
+                    enumClass != null ? (T[]) Array.newInstance(enumClass, numEnumConstants) : null;
+            String[] names = new String[numEnumConstants];
             for (int i = 0; i < numEnumConstants; i++) {
-                String enumName = in.readUTF();
+                names[i] = in.readUTF();
+                if (previousEnums == null) {
+                    continue;
+                }
                 try {
-                    previousEnums[i] = Enum.valueOf(enumClass, enumName);
+                    previousEnums[i] = Enum.valueOf(enumClass, names[i]);
                 } catch (IllegalArgumentException e) {
                     throw new IllegalStateException(
                             "Could not create a restore serializer for enum "
                                     + enumClass
                                     + ". 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);
         }
 

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java

@@ -48,10 +48,10 @@
 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;
 

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java

@@ -23,6 +23,7 @@
 import org.apache.flink.api.common.serialization.SerializerConfigImpl;
 import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil;
 import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil.IntermediateCompatibilityResult;
+import org.apache.flink.api.common.typeutils.CustomRestoreSerializerFactory;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
 import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
@@ -34,10 +35,12 @@
 
 import java.io.IOException;
 import java.lang.reflect.Field;
+import java.util.AbstractMap.SimpleEntry;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -143,6 +146,11 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
     @Override
     @SuppressWarnings("unchecked")
     public TypeSerializer<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 TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
         return TypeSerializerSchemaCompatibility.compatibleAsIs();
     }
 
+    // ---------------------------------------------------------------------------------------------
+    //  Schema extraction support
+    // ---------------------------------------------------------------------------------------------
+
+    /**
+     * Returns {@code true} if the POJO class could be loaded from the classloader that read this
+     * snapshot. When {@code false}, {@link #restoreSerializer()} delegates to the serializer
+     * supplied via {@link CustomRestoreSerializerFactory} instead of building a {@link
+     * PojoSerializer}.
+     */
+    @Internal
+    public boolean isPojoClassAvailable() {
+        return snapshotData.getPojoClass() != null;
+    }
+
+    /**
+     * Returns the POJO class name as stored in the snapshot. Available even when the class cannot
+     * be loaded.
+     */
+    @Internal
+    public String getPojoClassName() {
+        return snapshotData.getPojoClassName();
+    }
+
+    /**
+     * Returns an ordered list of (field name, field serializer snapshot) pairs. Field names are
+     * always present; snapshot values may be {@code null} when the field snapshot could not be
+     * read.
+     */
+    @Internal
+    public List<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
     // ---------------------------------------------------------------------------------------------