[FLINK-40298][table] Opt-in name-based state schema evolution for RowData

docs/layouts/shortcodes/generated/execution_config_configuration.html

@@ -356,6 +356,12 @@
             <td>Boolean</td>
             <td>Whether to compress spilled data. Currently we only support compress spilled data for sort and hash-agg and hash-join operators.</td>
         </tr>
+        <tr>
+            <td><h5>table.exec.state.schema-evolution.enabled</h5><br> <span class="label label-primary">Streaming</span></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>When enabled, a RowData keyed-state value whose schema changed in a backward-compatible way (adding nullable fields, reordering fields by name, evolving a nested ROW) is migrated when state is restored, instead of the restore failing. Only a state's own value serializer is covered: a RowData nested below a composite serializer, or a state whose serializer the operator pre-builds, is still rejected. Takes effect only on a state backend that migrates restored values at the object level, currently RocksDB; on other backends the restore still fails. Disabled by default.</td>
+        </tr>
         <tr>
             <td><h5>table.exec.state.ttl</h5><br> <span class="label label-primary">Streaming</span></td>
             <td style="word-wrap: break-word;">0 ms</td>

flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.serialization;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
 import org.apache.flink.configuration.PipelineOptions;
@@ -71,6 +72,21 @@ public interface SerializerConfig extends Serializable {
     /** Returns whether forces Flink to register Apache Avro classes in Kryo serializer. */
     TernaryBoolean isForceKryoAvroEnabled();
 
+    /**
+     * Whether state schema evolution for {@code RowData} state is enabled. When enabled, a {@code
+     * RowData} state value serializer admits a backward-compatible schema change and migrates the
+     * stored values when state is restored, instead of the restore failing.
+     *
+     * <p>This is a runtime hook rather than user-facing configuration. The option itself is owned
+     * by {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED} and only mirrored
+     * here by key, because this module cannot depend on the table API. Set the option there, not
+     * through this method.
+     */
+    @Internal
+    default boolean isStateSchemaEvolutionEnabled() {
+        return false;
+    }
+
     /**
      * Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link
      * PipelineOptions#FORCE_KRYO}.

flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java

@@ -23,6 +23,8 @@
 import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.PipelineOptions;
@@ -47,6 +49,14 @@ public final class SerializerConfigImpl implements SerializerConfig {
 
     private static final long serialVersionUID = 1L;
 
+    // Mirrors org.apache.flink.table.api.config.ExecutionConfigOptions
+    //   .TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED by key string, because flink-core cannot depend
+    //   on flink-table-api-java. The value travels in the shared job Configuration.
+    private static final ConfigOption<Boolean> STATE_SCHEMA_EVOLUTION_ENABLED =
+            ConfigOptions.key("table.exec.state.schema-evolution.enabled")
+                    .booleanType()
+                    .defaultValue(false);
+
     private final Configuration configuration;
 
     // ------------------------------- User code values --------------------------------------------
@@ -259,6 +269,11 @@ public void setForceKryo(boolean forceKryo) {
         configuration.set(PipelineOptions.FORCE_KRYO, forceKryo);
     }
 
+    @Override
+    public boolean isStateSchemaEvolutionEnabled() {
+        return configuration.get(STATE_SCHEMA_EVOLUTION_ENABLED);
+    }
+
     /** Returns whether the Apache Avro is the serializer for POJOs. */
     public boolean isForceAvroEnabled() {
         return configuration.get(PipelineOptions.FORCE_AVRO);
@@ -357,6 +372,10 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) {
         configuration
                 .getOptional(PipelineOptions.SERIALIZATION_CONFIG)
                 .ifPresent(c -> parseSerializationConfigWithExceptionHandling(classLoader, c));
+        configuration
+                .getOptional(STATE_SCHEMA_EVOLUTION_ENABLED)
+                .ifPresent(
+                        enabled -> this.configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, enabled));
     }
 
     @SuppressWarnings("unchecked")

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

@@ -0,0 +1,87 @@
+/*
+ * 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.apache.flink.api.common.functions.SerializerFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+/**
+ * A {@link TypeSerializer} that can admit a backward-compatible change of the schema of the values
+ * it serializes, provided the state it belongs to will actually have those values migrated.
+ *
+ * <p>Implementations are inert by default: a serializer only starts admitting schema changes once
+ * {@link #withStateSchemaEvolution()} has been called on it, and only if the job configuration
+ * opted in.
+ *
+ * @param <T> the type of the serialized values
+ */
+@Internal
+public interface StateSchemaEvolvingSerializer<T> {
+
+    /**
+     * Returns a serializer that admits a backward-compatible schema change and migrates the stored
+     * values when state is restored, or {@code this} if the job configuration did not opt in.
+     *
+     * <p>This is called only on a state's own value serializer, never on an arbitrary serializer
+     * encountered while walking a type.
+     */
+    TypeSerializer<T> withStateSchemaEvolution();
+
+    /**
+     * Decorates a factory so that the serializer it produces for a state value is armed for schema
+     * evolution.
+     *
+     * <p>Only a caller whose backend migrates restored values through {@link
+     * TypeSerializerSnapshot#migrate} may use this. A caller that does not decorate its factory
+     * keeps today's behavior unchanged.
+     */
+    static SerializerFactory arming(SerializerFactory delegate) {
+        // Not a lambda: SerializerFactory's single method is generic, which a lambda cannot
+        // implement.
+        return new SerializerFactory() {
+            @Override
+            public <T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation) {
+                return armStateValueSerializer(delegate.createSerializer(typeInformation));
+            }
+        };
+    }
+
+    /**
+     * Arms the serializer a state holds for its values, if it supports schema evolution at all.
+     *
+     * <p>The serializer passed in is armed, and nothing below it: there is no descent into a
+     * composite, and no recursion. A serializer nested below the state value is not reached by
+     * {@link TypeSerializerSnapshot#migrate}, so arming one would let a compatibility check report
+     * {@code compatibleAfterMigration} for bytes that nothing ever migrates.
+     *
+     * <p>The serializers armed here are therefore a subset of those {@code
+     * TtlAwareSerializer#wrapTtlAwareSerializer} descends into, which are the ones some backend
+     * calls {@code migrate} on. Being a subset is what keeps this sound. Widening the descent to
+     * close the gap is only safe once every caller registering the widened shape is known to reach
+     * such a backend, which is not true of the seam as it stands: operator state and broadcast
+     * state register list and map descriptors through it and never migrate.
+     */
+    @SuppressWarnings("unchecked")
+    static <T> TypeSerializer<T> armStateValueSerializer(TypeSerializer<T> serializer) {
+        return serializer instanceof StateSchemaEvolvingSerializer
+                ? ((StateSchemaEvolvingSerializer<T>) serializer).withStateSchemaEvolution()
+                : serializer;
+    }
+}

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

@@ -134,6 +134,41 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
     TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
             TypeSerializerSnapshot<T> oldSerializerSnapshot);
 
+    /**
+     * Migrates a single state value from the schema described by {@code oldSerializerSnapshot} to
+     * the schema described by this (new) snapshot. Like {@link
+     * #resolveSchemaCompatibility(TypeSerializerSnapshot)}, this is invoked on the new snapshot and
+     * receives the old snapshot as its argument.
+     *
+     * <p>The default implementation returns the value unchanged: a value already deserialized with
+     * the prior serializer is structurally compatible with the current serializer, so the caller
+     * can re-serialize it as-is. A serializer whose in-memory representation is coupled to its
+     * schema should override this to transform the value into the new layout -- for example by
+     * inserting nulls for added fields or reordering fields by name. An implementation may return
+     * the given value or a new instance.
+     *
+     * <p>The migration is not applied recursively to nested serializers. The snapshot of a
+     * composite type returns its value unchanged unless it overrides this method to decompose the
+     * value and migrate each part, so a caller that needs a nested value migrated must reach the
+     * nested snapshot itself. An implementation that does so should not assume that the old and the
+     * new snapshot expose nested snapshots of the same type: restoring may have replaced those of
+     * the old snapshot with decorators, so nested snapshots are best matched by position or name
+     * rather than by class.
+     *
+     * @param oldSerializerSnapshot snapshot of the serializer that wrote the value. A caller that
+     *     holds the snapshot persisted with the state should pass that one in preference to a
+     *     snapshot re-derived from a serializer restored from it, because that round trip does not
+     *     always reproduce the schema that was written.
+     * @param value the value, already deserialized with the prior serializer. It may be {@code
+     *     null} wherever the prior serializer can produce {@code null}. An implementation that
+     *     decomposes a composite value may likewise pass {@code null} to a nested snapshot for an
+     *     absent part, even where that part's serializer would reject {@code null} at top level.
+     * @return the value adapted to the schema of the current serializer.
+     */
+    default T migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value) {
+        return value;
+    }
+
     // ------------------------------------------------------------------------
     //  read / write utilities
     // ------------------------------------------------------------------------

flink-core/src/test/java/org/apache/flink/api/common/serialization/SerializerConfigImplTest.java

@@ -20,6 +20,8 @@
 
 import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 
 import com.esotericsoftware.kryo.Kryo;
@@ -42,6 +44,47 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class SerializerConfigImplTest {
+
+    /**
+     * Mirrors {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED}, which lives
+     * in a module this one cannot depend on.
+     */
+    private static final ConfigOption<Boolean> STATE_SCHEMA_EVOLUTION_ENABLED =
+            ConfigOptions.key("table.exec.state.schema-evolution.enabled")
+                    .booleanType()
+                    .defaultValue(false);
+
+    @Test
+    void testStateSchemaEvolutionFlagSurvivesConstructionAndCopy() {
+        Configuration configuration = new Configuration();
+        configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, true);
+
+        SerializerConfigImpl serializerConfig = new SerializerConfigImpl(configuration);
+
+        assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isTrue();
+        assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isTrue();
+    }
+
+    @Test
+    void testStateSchemaEvolutionFlagDefaultsFalseThroughCopy() {
+        SerializerConfigImpl serializerConfig = new SerializerConfigImpl();
+
+        assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isFalse();
+        assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isFalse();
+    }
+
+    @Test
+    void testStateSchemaEvolutionFlagSurvivesConfigure() {
+        Configuration configuration = new Configuration();
+        configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, true);
+
+        SerializerConfigImpl serializerConfig = new SerializerConfigImpl();
+        serializerConfig.configure(configuration, SerializerConfigImplTest.class.getClassLoader());
+
+        assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isTrue();
+        assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isTrue();
+    }
+
     @Test
     void testReadingDefaultConfig() {
         SerializerConfig config = new SerializerConfigImpl();

flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java

@@ -54,6 +54,15 @@ public TypeSerializerSchemaCompatibility<Integer> resolveSchemaCompatibility(
                 .isTrue();
     }
 
+    @Test
+    void testMigrateReturnsValueUnchangedByDefault() {
+        TypeSerializerSnapshot<Integer> oldSnapshot = new NotCompletedTypeSerializerSnapshot();
+        TypeSerializerSnapshot<Integer> newSnapshot = new NotCompletedTypeSerializerSnapshot();
+        Integer value = 1000;
+
+        assertThat(newSnapshot.migrate(oldSnapshot, value)).isSameAs(value);
+    }
+
     private static class NotCompletedTypeSerializer extends TypeSerializer<Integer> {
 
         @Override

flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java

@@ -20,9 +20,12 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.SerializerFactory;
 import org.apache.flink.api.common.state.InternalCheckpointListener;
 import org.apache.flink.api.common.state.State;
 import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.core.fs.CloseableRegistry;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
@@ -387,7 +390,7 @@ public <N, S extends State, V> S getOrCreateKeyedState(
         InternalKvState<K, ?, ?> kvState = keyValueStatesByName.get(stateDescriptor.getName());
         if (kvState == null) {
             if (!stateDescriptor.isSerializerInitialized()) {
-                stateDescriptor.initializeSerializerUnlessSet(executionConfig);
+                stateDescriptor.initializeSerializerUnlessSet(stateValueSerializerFactory());
             }
             kvState =
                     MetricsTrackingStateFactory.createStateAndWrapWithMetricsTrackingIfEnabled(
@@ -403,6 +406,29 @@ public <N, S extends State, V> S getOrCreateKeyedState(
         return (S) kvState;
     }
 
+    /**
+     * The factory a state descriptor registered here uses for its own value serializer. It
+     * reproduces what {@link StateDescriptor#initializeSerializerUnlessSet(ExecutionConfig)}
+     * builds, and arms schema evolution on top of it only when this backend migrates restored
+     * values at the object level.
+     */
+    private SerializerFactory stateValueSerializerFactory() {
+        SerializerFactory factory =
+                new SerializerFactory() {
+                    @Override
+                    public <T> TypeSerializer<T> createSerializer(
+                            TypeInformation<T> typeInformation) {
+                        return typeInformation.createSerializer(
+                                executionConfig == null
+                                        ? null
+                                        : executionConfig.getSerializerConfig());
+                    }
+                };
+        return supportsObjectLevelValueMigration()
+                ? StateSchemaEvolvingSerializer.arming(factory)
+                : factory;
+    }
+
     public void publishQueryableStateIfEnabled(
             StateDescriptor<?, ?> stateDescriptor, InternalKvState<?, ?, ?> kvState) {
         if (stateDescriptor.isQueryable()) {