From 9a3e578ff0908c1741ec30e2d179b1defe1f9921 Mon Sep 17 00:00:00 2001
From: Weiqing Yang <yangweiqing001@gmail.com>
Date: Sun, 2 Aug 2026 17:23:24 -0700
Subject: [PATCH] [FLINK-40296][core] Add object-level migrate hook to
 TypeSerializerSnapshot

Add a default method that lets a serializer snapshot transform an already
deserialized state value from the schema it was written with into the schema
the current serializer expects:

    default T migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value)

Like resolveSchemaCompatibility, it is invoked on the new snapshot and receives
the old snapshot as its argument. The default returns the value unchanged, so
behavior is unaffected for every existing serializer: a value deserialized with
the prior serializer is structurally compatible with the current one and can be
re-serialized as is.

The javadoc states that migration is not applied recursively to nested
serializers. Unlike resolveSchemaCompatibility, which CompositeTypeSerializer-
Snapshot delegates to the nested snapshots, migrate has no delegating override,
so a composite returns its value unmigrated unless it decomposes the value
itself. That asymmetry is invisible at the call site and would otherwise fail
silently.

Generated-by: Claude Code (Opus 5)
---
 .../typeutils/TypeSerializerSnapshot.java     | 26 +++++++++++++++++++
 .../typeutils/TypeSerializerSnapshotTest.java |  9 +++++++
 2 files changed, 35 insertions(+)

diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
index 1fe4134ee51c9a..e4788736a68a1f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java
@@ -134,6 +134,32 @@ 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.
+     *
+     * @param oldSerializerSnapshot snapshot of the serializer that wrote the value.
+     * @param value the value, already deserialized with the prior serializer.
+     * @return the value adapted to the schema of the current serializer.
+     */
+    default T migrate(TypeSerializerSnapshot<T> oldSerializerSnapshot, T value) {
+        return value;
+    }
+
     // ------------------------------------------------------------------------
     //  read / write utilities
     // ------------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java
index b176adc482b36f..3dbace1bf58583 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java
+++ b/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