acrites commented on code in PR #40750:
URL: https://github.com/apache/beam/pull/40750#discussion_r4391659720
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/state/WindmillMapViaMultimap.java:
##########
@@ -62,19 +62,26 @@ public void put(KeyT key, ValueT value) {
multimap.put(key, value);
}
+ @SuppressWarnings("nullness")
@Override
public ReadableState<ValueT> computeIfAbsent(
KeyT key, Function<? super KeyT, ? extends ValueT> mappingFunction) {
// Note that computeIfAbsent comments indicate that the read is lazy but
this matches the
- // existing eager
- // behavior of WindmillMap.
+ // existing eager behavior of WindmillMap.
Iterable<ValueT> existingValues = multimap.get(key).read();
- if (Iterables.isEmpty(existingValues)) {
+ Iterator<ValueT> iterator = existingValues.iterator();
+ if (!iterator.hasNext()) {
ValueT inserted = mappingFunction.apply(key);
multimap.put(key, inserted);
return ReadableStates.immediate(inserted);
} else {
- return
ReadableStates.immediate(Iterables.getOnlyElement(existingValues));
+ ValueT existing = Iterators.getOnlyElement(iterator);
+ if (existing == null) {
Review Comment:
Some other implementations I found were `InMemoryMap`:
```
public ReadableState<V> computeIfAbsent(
K key, Function<? super K, ? extends V> mappingFunction) {
V v = contents.get(key);
if (v == null) {
v = contents.put(key, mappingFunction.apply(key));
}
return ReadableStates.immediate(v);
}
```
Here is the Spark runner's implementation (note it's return value matches
the original version of `MapState.computeIfAbsent()`):
```
public ReadableState<MapValueT> computeIfAbsent(
MapKeyT key, Function<? super MapKeyT, ? extends MapValueT>
mappingFunction) {
Map<MapKeyT, MapValueT> sparkMapState = readAsMap();
MapValueT current = sparkMapState.get(key);
if (current == null) {
put(key, mappingFunction.apply(key));
}
return ReadableStates.immediate(current);
}
```
It looks like the Flink runner has the same implementation as the Spark
runner:
```
public ReadableState<ValueT> computeIfAbsent(
final KeyT key, Function<? super KeyT, ? extends ValueT>
mappingFunction) {
try {
ValueT current =
flinkStateBackend
.getPartitionedState(namespace, namespaceSerializer,
flinkStateDescriptor)
.get(key);
if (current == null) {
flinkStateBackend
.getPartitionedState(namespace, namespaceSerializer,
flinkStateDescriptor)
.put(key, mappingFunction.apply(key));
}
return ReadableStates.immediate(current);
} catch (Exception e) {
throw new RuntimeException("Error put kv to state.", e);
}
}
```
--
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]
