Copilot commented on code in PR #28973:
URL: https://github.com/apache/flink/pull/28973#discussion_r4310763816
##########
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:
The backend capability check only applies while creating a serializer. If a
`ValueStateDescriptor` is first registered on RocksDB and then reused in memory
on a heap backend, it retains its armed serializer: this initialization is
skipped, and `getSerializer().duplicate()` preserves the flag.
`DefaultKeyedStateStore` also leaves that cached serializer unchanged. Heap
accepts `compatibleAfterMigration` without invoking the row migration hook when
TTL is unchanged, so an evolved schema can be registered while values still
have the old binary layout. Reject already-armed serializers on unsupported
backends, or keep arming local to each registration rather than caching it in
the descriptor. Add a regression test that reuses the same descriptor across
the two backend capabilities.
##########
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:
This also opts anonymous accumulator layouts into name-based evolution.
`ProcTimeUnboundedPrecedingFunction.open()` builds its `accState` with
`InternalTypeInfo.ofFields(accTypes)`, which generates positional names `f0`,
`f1`, etc. With the option enabled on RocksDB, inserting a nullable INT MIN
accumulator before two existing INT MIN accumulators looks like appending `f2`.
Migration then puts the first old minimum into the new aggregate, the second
into the first existing aggregate, and null into the second existing aggregate.
Previously, the arity change rejected the restore. Keep these positional-only
accumulator layouts unarmed unless stable accumulator identities and supported
migration semantics are supplied, and test an insertion into an existing
accumulator layout.
--
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]
