rkhachatryan commented on code in PR #19679:
URL: https://github.com/apache/flink/pull/19679#discussion_r935538992
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapAggregatingState.java:
##########
@@ -136,18 +144,29 @@ public ACC apply(ACC accumulator, IN value) {
}
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
static <T, K, N, SV, S extends State, IS extends S> IS create(
StateDescriptor<S, SV> stateDesc,
StateTable<K, N, SV> stateTable,
- TypeSerializer<K> keySerializer) {
- return (IS)
- new HeapAggregatingState<>(
- stateTable,
- keySerializer,
- stateTable.getStateSerializer(),
- stateTable.getNamespaceSerializer(),
- stateDesc.getDefaultValue(),
- ((AggregatingStateDescriptor<T, SV, ?>)
stateDesc).getAggregateFunction());
+ TypeSerializer<K> keySerializer,
+ @Nullable IS existingState) {
+ return existingState == null
+ ? (IS)
+ new HeapAggregatingState<>(
Review Comment:
I think "centralized" means one function does one of two things, depending
on the input argument. I don't see any benefit from it.
If that's done by the client, then
- the logic wouldn't be duplicated in every `createOrUpdate`
- separate `create` and `update` functions are more clear than the combined
one
--
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]
