This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 57fd225b09c [FLINK-40296][core] Add object-level migrate hook to
TypeSerializerSnapshot (#28880)
57fd225b09c is described below
commit 57fd225b09cf61e204ea1d1c42a7d351c14c78d3
Author: Weiqing Yang <[email protected]>
AuthorDate: Mon Aug 24 07:31:09 2026 -0700
[FLINK-40296][core] Add object-level migrate hook to TypeSerializerSnapshot
(#28880)
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)
---
.../common/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 1fe4134ee51..e4788736a68 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 @@ public interface TypeSerializerSnapshot<T> {
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 b176adc482b..3dbace1bf58 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 @@ class TypeSerializerSnapshotTest {
.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
