weiqingy opened a new pull request, #28881: URL: https://github.com/apache/flink/pull/28881
This is the second PR of the [FLIP-527](https://cwiki.apache.org/confluence/spaces/FLINK/pages/353601981/FLIP-527+State+Schema+Evolution+for+RowData) implementation, split into a stack of small, independently reviewable PRs under the umbrella issue [FLINK-37732](https://issues.apache.org/jira/browse/FLINK-37732). Landing order: | Step | Sub-task | Scope | |---|---|---| | PR-1 | [FLINK-40296](https://issues.apache.org/jira/browse/FLINK-40296) | Object-level `migrate` hook on `TypeSerializerSnapshot` | | **PR-2 (this PR)** | [FLINK-40297](https://issues.apache.org/jira/browse/FLINK-40297) | Route TTL-aware value migration through the hook | | PR-3 | [FLINK-40298](https://issues.apache.org/jira/browse/FLINK-40298) | Opt-in name-based schema evolution for `RowData` | | PR-4 | [FLINK-40299](https://issues.apache.org/jira/browse/FLINK-40299) | End-to-end state migration coverage on RocksDB | Draft until PR-1 merges: this branch contains PR-1's commit, so the diff against master shows both. It will be rebased onto master and marked ready once PR-1 lands. PR-1 and PR-2 are behavior-neutral, since the hook defaults to returning its argument and nothing overrides it until PR-3. ## What is the purpose of the change `TtlAwareSerializer.migrateValueFromPriorSerializer` is the single entry point through which the RocksDB state backend migrates state values on restore. `AbstractRocksDBState`, `RocksDBListState` and `RocksDBMapState` all call it, each after unwrapping the state shape it owns: the value serializer, the list element serializer, or the map value serializer. Today it deserializes with the prior serializer and re-serializes with the new one, giving a serializer no opportunity to adapt the value in between. This routes it through the `migrate` hook added in PR-1, so a serializer that overrides the hook takes effect. Behavior is unchanged, because nothing overrides it yet and the default returns the value unchanged. Two details are worth calling out for review, because both are contract decisions rather than mechanics. **The hook receives the persisted prior snapshot, not a re-derived one.** The prior serializer reaching this method is itself `previousSerializerSnapshot.restoreSerializer()`, so calling `snapshotConfiguration()` on it is a snapshot to serializer to snapshot round trip. That round trip is lossy: `PojoSerializerSnapshot` inserts a `null` `Field` for a field that no longer exists on the class, and re-snapshotting substitutes a synthetic name for it. An override reconciling fields by name, which is what FLIP-527 adds in PR-3, would then see a fabricated schema. The backend already fetches the persisted snapshot above the migration loop, so this threads it down and each caller descends it alongside the serializer it already descends. **The descent unwraps the TtlAware decorator first, and an unexpected snapshot type now fails rather than falling back.** Registering a new serializer mutates the previous snapshot's nested snapshots in place, so a list or map state's persisted element or value snapshot is a `TtlAwareSerializerSnapshot` rather than the snapshot the checkpoint wrote. An earlier revision of this change assumed otherwise and fell back silently to the re-derived snapshot on exactly those paths; the assertion added here is what surfaced it. Only an absent snapshot falls back now, which is the case where nothing was persisted and the re-derived snapshot is all that exists. ## Brief change log - Unwrap the prior value to its bare, non-TTL form, pass it through `migrate`, and re-wrap when this serializer is TTL-enabled, preserving the prior TTL timestamp when the prior value carried one - Thread the persisted prior snapshot from the backend down to the hook, descending it alongside the serializer at each of the three call sites - Fail with a clear message when the persisted snapshot is of an unexpected type, instead of silently re-deriving one - Pin the nullability of the hook's `value` parameter, and note that old and new snapshots need not expose the same nested snapshot types ## Verifying this change This change added tests and can be verified as follows: - `TtlAwareSerializerTest` covers all four combinations of prior and current TTL being enabled, asserting migrated bytes and the preserved timestamp. These were written to pass against the pre-change code, so they demonstrate that behavior is unchanged rather than asserting it - Tests pin which snapshot reaches the hook, by object identity: a re-derived snapshot has the same class and a different identity, so identity is the only discriminator. Removing either the TtlAware unwrap or the TTL descent fails these - Tests assert the type guards fire, in both directions and behind the TtlAware layer. Softening either guard alone still fails - `StateSerializerProviderTest` pins the in-place mutation of nested snapshots that the descent depends on, so a change to that mechanism fails here rather than silently downstream - Covered end to end by the existing RocksDB state migration suites, which exercise this method through all three state shapes in both TTL directions ## 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)`: no. `TypeSerializerSnapshot` is `@PublicEvolving` but only its javadoc changed. Three internal backend method signatures widened; none is on an interface or carries a stability annotation, and japicmp is green. - The serializers: yes - The runtime per-record code paths (performance sensitive): no for the steady-state record path. This method runs once per state entry during restore, so the snapshot lookup is hoisted out of the loop and guard messages are built only on failure. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, this is on the checkpoint and savepoint restore path, though behavior is unchanged - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no, it is preparatory for FLIP-527 - If yes, how is the feature documented? JavaDocs --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Opus 5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
