[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/java/typeutils/runtime/PojoDeserializerCompatibilitySnapshot.java

@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.IOException;
+
+/**
+ * A {@link TypeSerializerSnapshot} for deserializers that can read POJO binary data without the
+ * user POJO class being on the classpath.
+ *
+ * <p>This snapshot declares itself {@link TypeSerializerSchemaCompatibility#compatibleAsIs()
+ * compatible as-is} with any stored {@link PojoSerializerSnapshot}. It is used by the {@code
+ * PojoToRowDataDeserializer} in the state processing API (which converts them to {@code
+ * GenericRowData}).
+ *
+ * <p>This snapshot only ever exists in-memory, wrapping the live deserializer it was created from:
+ * composite schema compatibility checks (e.g. {@code
+ * CompositeTypeSerializerSnapshot#resolveOuterSchemaCompatibility}) restore the "new" side of a
+ * composite serializer (such as a timer serializer nesting the key serializer) to compare it
+ * against the old one, even when nested-level compatibility already short-circuited to {@link
+ * TypeSerializerSchemaCompatibility#compatibleAsIs()}. {@link #restoreSerializer()} therefore
+ * returns the wrapped instance instead of throwing, since it never actually needs to be
+ * reconstructed from a persisted snapshot.
+ */
+@Internal
+public final class PojoDeserializerCompatibilitySnapshot<T> implements TypeSerializerSnapshot<T> {
+
+    private final TypeSerializer<T> restoredSerializer;
+
+    public PojoDeserializerCompatibilitySnapshot() {
+        this.restoredSerializer = 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) {
+            return restoredSerializer;
+        }
+        throw new UnsupportedOperationException(
+                "PojoDeserializerCompatibilitySnapshot cannot reconstruct the deserializer on its own. "
+                        + "Use PojoSerializerSnapshot.restoreSerializer() or "
+                        + "PojoToRowDataDeserializer.create().");
+    }
+
+    @Override
+    public void writeSnapshot(DataOutputView out) throws IOException {}
+
+    @Override
+    public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+            throws IOException {}
+}

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

@@ -54,6 +54,42 @@ public class PojoSerializerSnapshot<T> implements TypeSerializerSnapshot<T> {
      */
     private static final int VERSION = 2;
 
+    /**
+     * Optional thread-local factory for building a custom serializer when the POJO class is not on
+     * the classpath.
+     *
+     * <p>Used by the State Processing API to inject {@code PojoToRowDataDeserializer} so that the
+     * heap state backend can restore POJO state without the user class (values are converted to
+     * {@code GenericRowData} during restore instead of raising {@link ClassNotFoundException}).
+     *
+     * <p>Set via {@link #setPojoRestoreSerializerFactory} before the state backend restore and
+     * cleared via {@link #clearPojoRestoreSerializerFactory} afterward.
+     */
+    private static final ThreadLocal<
+                    java.util.function.Function<PojoSerializerSnapshot<?>, TypeSerializer<?>>>
+            RESTORE_FACTORY = new ThreadLocal<>();
+
+    /**
+     * Registers a factory that provides a fallback {@link TypeSerializer} when the POJO class is
+     * absent from the classpath.
+     *
+     * <p>The factory receives the snapshot and should return a serializer that can read the POJO
+     * binary format.
+     *
+     * <p>Must be paired with {@link #clearPojoRestoreSerializerFactory()} in a try-finally block.
+     */
+    @Internal
+    public static void setPojoRestoreSerializerFactory(
+            java.util.function.Function<PojoSerializerSnapshot<?>, TypeSerializer<?>> factory) {
+        RESTORE_FACTORY.set(factory);
+    }
+
+    /** Clears the factory registered via {@link #setPojoRestoreSerializerFactory}. */
+    @Internal
+    public static void clearPojoRestoreSerializerFactory() {
+        RESTORE_FACTORY.remove();
+    }
+
     /** Contains the actual content for the serializer snapshot. */
     private PojoSerializerSnapshotData<T> snapshotData;
 
@@ -143,6 +179,15 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
     @Override
     @SuppressWarnings("unchecked")
     public TypeSerializer<T> restoreSerializer() {
+        if (snapshotData.getPojoClass() == null) {
+            java.util.function.Function<PojoSerializerSnapshot<?>, TypeSerializer<?>> factory =
+                    RESTORE_FACTORY.get();
+            if (factory != null) {
+                return (TypeSerializer<T>) factory.apply(this);
+            }
+            throw new RuntimeException(new ClassNotFoundException(snapshotData.getPojoClassName()));
+        }
+
         final int numFields = snapshotData.getFieldSerializerSnapshots().size();
 
         final ArrayList<Field> restoredFields = new ArrayList<>(numFields);
@@ -257,6 +302,73 @@ public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
         return TypeSerializerSchemaCompatibility.compatibleAsIs();
     }
 
+    // ---------------------------------------------------------------------------------------------
+    //  Schema extraction support
+    // ---------------------------------------------------------------------------------------------
+
+    /**
+     * Returns the raw snapshot data. Exposed for schema extraction utilities that need to inspect
+     * field names and field serializer snapshots without the user POJO class being on the
+     * classpath.
+     *
+     * @return the snapshot data; field names are always present, field serializer snapshots may be
+     *     null for individual fields if their snapshot could not be read
+     */
+    @Internal
+    public PojoSerializerSnapshotData<T> getSnapshotData() {
+        return snapshotData;
+    }
+
+    /**
+     * Returns {@code true} if the POJO class can be loaded from the current classloader.
+     *
+     * <p>When {@code false}, {@link #restoreSerializer()} returns a (or a factory-supplied
+     * deserializer) instead of the normal {@link PojoSerializer}.
+     */
+    @Internal
+    public boolean isPojoClassAvailable() {
+        return snapshotData.getPojoClass() != null;
+    }
+
+    /**
+     * Returns the POJO class name as stored in the snapshot. Available even when the class cannot
+     * be loaded.
+     */
+    @Internal
+    public String getPojoClassName() {
+        return snapshotData.getPojoClassName();
+    }
+
+    /**
+     * Returns an ordered list of (field name, field TypeSerializerSnapshot) pairs. Field names are
+     * always present; snapshot values may be {@code null} when the field snapshot could not be
+     * read.
+     */
+    @Internal
+    public java.util.List<java.util.AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>>
+            getFieldSnapshotEntries() {
+        java.util.List<java.util.AbstractMap.SimpleEntry<String, TypeSerializerSnapshot<?>>>
+                result = new java.util.ArrayList<>();
+        snapshotData
+                .getFieldSerializerSnapshots()
+                .forEach(
+                        (fieldName, field, fieldSnapshot) ->
+                                result.add(
+                                        new java.util.AbstractMap.SimpleEntry<>(
+                                                fieldName, fieldSnapshot)));
+        return result;
+    }
+
+    /**
+     * Returns the registered subclass serializer snapshots in tag order (tag 0, 1, 2, …). Values
+     * may be {@code null} if a subclass snapshot was not readable.
+     */
+    @Internal
+    public java.util.List<TypeSerializerSnapshot<?>> getRegisteredSubclassSnapshotsOrdered() {
+        return new java.util.ArrayList<>(
+                snapshotData.getRegisteredSubclassSerializerSnapshots().unwrapOptionals().values());
+    }
+
     // ---------------------------------------------------------------------------------------------
     //  Utility methods
     // ---------------------------------------------------------------------------------------------

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

@@ -24,14 +24,15 @@
 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;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.LinkedHashMap;
@@ -113,6 +114,7 @@ static <T> PojoSerializerSnapshotData<T> createFrom(
 
         return new PojoSerializerSnapshotData<>(
                 pojoClass,
+                pojoClass.getName(),
                 fieldSerializerSnapshots,
                 optionalMapOf(registeredSubclassSerializerSnapshots, Class::getName),
                 optionalMapOf(nonRegisteredSubclassSerializerSnapshots, Class::getName));
@@ -153,27 +155,31 @@ static <T> PojoSerializerSnapshotData<T> createFrom(
 
         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;
     private LinkedOptionalMap<Class<?>, TypeSerializerSnapshot<?>>
             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 +192,7 @@ private PojoSerializerSnapshotData(
     // ---------------------------------------------------------------------------------------------
 
     void writeSnapshotData(DataOutputView out) throws IOException {
-        out.writeUTF(pojoClass.getName());
+        out.writeUTF(pojoClassName);
         writeOptionalMap(
                 out,
                 fieldSerializerSnapshots,
@@ -206,7 +212,17 @@ void writeSnapshotData(DataOutputView out) throws IOException {
 
     private static <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 = null;
+        try {
+            @SuppressWarnings("unchecked")
+            Class<T> resolved = (Class<T>) Class.forName(pojoClassName, false, userCodeClassLoader);
+            pojoClass = resolved;
+        } catch (ClassNotFoundException e) {
+            LOG.debug(
+                    "POJO class '{}' not found on classpath; schema can still be read from field snapshots.",
+                    pojoClassName);
+        }
 
         LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> fieldSerializerSnapshots =
                 readOptionalMap(
@@ -226,6 +242,7 @@ private static <T> PojoSerializerSnapshotData<T> readSnapshotData(
 
         return new PojoSerializerSnapshotData<>(
                 pojoClass,
+                pojoClassName,
                 fieldSerializerSnapshots,
                 registeredSubclassSerializerSnapshots,
                 nonRegisteredSubclassSerializerSnapshots);
@@ -235,10 +252,15 @@ private static <T> PojoSerializerSnapshotData<T> readSnapshotData(
     //  Snapshot data accessors
     // ---------------------------------------------------------------------------------------------
 
+    @Nullable
     Class<T> getPojoClass() {
         return pojoClass;
     }
 
+    String getPojoClassName() {
+        return pojoClassName;
+    }
+
     LinkedOptionalMap<Field, TypeSerializerSnapshot<?>> getFieldSerializerSnapshots() {
         return fieldSerializerSnapshots;
     }