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 1fe4134ee51c9..e4788736a68a1 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 resolveSchemaCompatibility( TypeSerializerSnapshot 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. + * + *

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. + * + *

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 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 b176adc482b36..3dbace1bf5858 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 resolveSchemaCompatibility( .isTrue(); } + @Test + void testMigrateReturnsValueUnchangedByDefault() { + TypeSerializerSnapshot oldSnapshot = new NotCompletedTypeSerializerSnapshot(); + TypeSerializerSnapshot newSnapshot = new NotCompletedTypeSerializerSnapshot(); + Integer value = 1000; + + assertThat(newSnapshot.migrate(oldSnapshot, value)).isSameAs(value); + } + private static class NotCompletedTypeSerializer extends TypeSerializer { @Override