masteryhx commented on code in PR #19679:
URL: https://github.com/apache/flink/pull/19679#discussion_r906965060
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java:
##########
@@ -477,6 +478,50 @@ KeyGroupedInternalPriorityQueue<T> create(
}
}
+ @Override
+ public <N, S extends State, V> S upgradeKeyedState(
+ TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V>
stateDescriptor)
+ throws Exception {
+ StateFactory stateFactory = getStateFactory(stateDescriptor);
+ Tuple2<ColumnFamilyHandle, RegisteredKeyValueStateBackendMetaInfo<N,
V>> registerResult =
+ tryRegisterKvStateInformation(stateDescriptor,
namespaceSerializer, noTransform());
+
Preconditions.checkState(kvStateInformation.containsKey(stateDescriptor.getName()));
+ kvStateInformation.computeIfPresent(
+ stateDescriptor.getName(),
+ (stateName, kvStateInfo) ->
+ new RocksDbKvStateInfo(
+ kvStateInfo.columnFamilyHandle,
+ new RegisteredKeyValueStateBackendMetaInfo<>(
+ kvStateInfo.metaInfo.snapshot())));
+ return stateFactory.createState(
+ stateDescriptor, registerResult,
RocksDBKeyedStateBackend.this);
Review Comment:
If we wanna to create the object only once, not only we need to define a
method like RocksDbValueState.withNewSerializer(...) as roman said,
but also we need to store the created InternalKvState in
RocksDBKeyedStateBackend / HashMapKeyedStateBackend, and modify the existing
logic of createInternalState (to check whether there are created
InternalKvState).
I think it could work to avoid something unexpected while creating the
object more than once as roman said.
But it does need to modify some existing logic, so maybe @Myasuka could
share some ideas about it ?
--
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]
