[FLINK-40296][core] Add object-level migrate hook to TypeSerializerSnapshot

[weiqingy]

This is the first PR of the FLIP-527 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-37732. Landing order:

Step	Sub-task	Scope
PR-1 (this PR)	FLINK-40296	Object-level migrate hook on TypeSerializerSnapshot
PR-2	FLINK-40297	Route TTL-aware value migration through the hook
PR-3	FLINK-40298	Opt-in name-based schema evolution for RowData
PR-4	FLINK-40299	End-to-end state migration coverage on RocksDB

Each PR depends on the one before it. PR-1 and PR-2 are behavior-neutral: the hook added here defaults to returning its argument, so no existing serializer changes behavior until PR-3 overrides it for RowData.

What is the purpose of the change

FLIP-527 makes a RowData state value survive a backward-compatible schema change, such as a nullable field appended by an upstream Avro schema. Doing that needs a way for a serializer snapshot to transform an already deserialized value from the schema it was written with into the schema the current serializer expects. RowData needs this because its binary layout is fixed-width and position-addressed, so re-serializing a value read with the old layout does not re-pack it.

This PR adds only that extension point:

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, which is correct for every serializer whose in-memory representation does not depend on the schema: such a value is structurally compatible with the current serializer and can be re-serialized as is.

Nothing calls the hook yet. The caller lands in PR-2 and the first override in PR-3. Keeping the interface change on its own keeps the public extension point reviewable in isolation.

Brief change log

• Add a default migrate method to TypeSerializerSnapshot, returning the value unchanged
• Document that migration is not applied recursively to nested serializers: unlike resolveSchemaCompatibility, which CompositeTypeSerializerSnapshot delegates to the nested snapshots, migrate has no delegating override, so a composite returns its value unmigrated unless it decomposes the value itself

Verifying this change

This change added tests and can be verified as follows:

• TypeSerializerSnapshotTest#testMigrateReturnsValueUnchangedByDefault asserts that a snapshot which does not override migrate returns the same instance it was given. The identity assertion is deliberate rather than an equality one: a default that copied would be a silent per value allocation on the restore path.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes, TypeSerializerSnapshot is @PublicEvolving. The addition is a default method, so it is source and binary compatible and no existing implementor needs to change.
• The serializers: yes, it adds a method to the serializer snapshot interface, though no serializer behavior changes in this PR
• The runtime per-record code paths (performance sensitive): no, the method has no caller in this PR, and its eventual caller is on the restore path rather than the steady-state record path
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes, it is the first step of FLIP-527
• If yes, how is the feature documented? JavaDocs. The user-facing option and its documentation land in PR-3.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5)

[flinkbot]

CI report:

• 9a3e578 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build