weiqingy commented on code in PR #28973:
URL: https://github.com/apache/flink/pull/28973#discussion_r4318185516
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java:
##########
@@ -387,7 +390,7 @@ public <N, S extends State, V> S getOrCreateKeyedState(
InternalKvState<K, ?, ?> kvState =
keyValueStatesByName.get(stateDescriptor.getName());
if (kvState == null) {
if (!stateDescriptor.isSerializerInitialized()) {
- stateDescriptor.initializeSerializerUnlessSet(executionConfig);
+
stateDescriptor.initializeSerializerUnlessSet(stateValueSerializerFactory());
Review Comment:
Valid, fixed in dafdbc2d323: keyed backends without object-level value
migration now reject a descriptor whose cached serializer is already armed,
with a `StateMigrationException` asking for a separate descriptor instance. The
check is in `AbstractKeyedStateBackend.getOrCreateKeyedState`, which covers
heap, ForSt sync and the `DefaultKeyedStateStore` path, and in
`ChangelogKeyedStateBackend.getOrCreateKeyedState`, since the changelog backend
creates states on the backend it wraps directly. Both run before TTL wrapping,
so they see the descriptor's own serializer.
I chose rejecting over per-registration arming: the descriptor caches its
serializer once and has no copy or reset path, so arming per registration would
change `StateDescriptor` caching in flink-core for every backend, while
rejecting fails closed. The reverse order needs no check: a descriptor first
used on a non-migrating backend keeps an unarmed serializer, so on RocksDB it
behaves as with the option off.
Tests that reuse one descriptor across both capabilities:
-
`StateSchemaEvolutionArmingTest#descriptorArmedOnAMigratingBackendIsRejectedWithoutObjectLevelMigration`,
plus its TTL variant
-
`#descriptorArmedThroughTheKeyedStoreIsRejectedWithoutObjectLevelMigration`
(the `DefaultKeyedStateStore` path)
-
`#descriptorFirstUsedWithoutObjectLevelMigrationStaysUnarmedOnAMigratingBackend`
(reverse order)
-
`ChangelogKeyedStateBackendSchemaEvolutionTest#descriptorArmedOnAMigratingBackendIsRejected`
(changelog over a real heap backend)
ForSt async (v2 descriptors, never armed) and batch (never snapshots or
restores keyed state) need no check.
##########
flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalTypeInfo.java:
##########
@@ -194,7 +194,19 @@ public boolean isKeyType() {
}
@Override
+ @SuppressWarnings("unchecked")
public TypeSerializer<T> createSerializer(SerializerConfig config) {
+ if (config != null
+ && config.isStateSchemaEvolutionEnabled()
+ && typeSerializer instanceof RowDataSerializer) {
Review Comment:
Valid, fixed in d5166f87874: type information built by
`InternalTypeInfo.ofFields(LogicalType...)` no longer opts in, since its
positional `f0..fn` names cannot tell an inserted field from an appended one.
These layouts keep master's behaviour, so a change in the number or types of
fields is rejected at restore. The mark is set by that factory, not inferred
from the names. It is kept across Java serialization, since a descriptor may be
shipped before its serializer is created, and it is not part of
`equals`/`hashCode`.
This covers the aggregate and OVER accumulator value states:
`GroupAggFunction`, `MiniBatchGroupAggFunction`,
`MiniBatchGlobalGroupAggFunction`, `GroupTableAggFunction` and the OVER
functions, including `ProcTimeUnboundedPrecedingFunction`. Window, data-view
and async-state accumulators were never armed (pre-built serializers or v2
descriptors). The option description now lists aggregate and OVER accumulator
state among the paths it does not cover.
Tests:
-
`RowDataStateSchemaEvolutionArmingTest#insertionIntoPositionalLayoutIsIncompatible`
is the comment's example, an INT inserted before two INTs, resolved against
both a named and a name-less old snapshot.
-
`ProcTimeUnboundedPrecedingFunctionTest#accumulatorInsertionIsRejectedWithSchemaEvolutionEnabled`
restores on RocksDB with the option on after inserting a same-typed
accumulator before the existing one, and expects the restore to fail with a
`StateMigrationException`.
Evolving these layouts would need stable accumulator identities from the
planner and migration semantics for them, which is out of scope for this PR.
--
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]
