rkhachatryan commented on code in PR #29064:
URL: https://github.com/apache/flink/pull/29064#discussion_r3927776380
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java:
##########
@@ -221,7 +225,48 @@ private static <T> GateFilterHandler<T> createGateHandler(
return null;
}
- return new GateFilterHandler<>(gateVirtualChannels, elementSerializer);
+ return new GateFilterHandler<>(gateVirtualChannels, elementSerializer,
channelMapping);
+ }
+
+ /**
+ * Maps each old-channel key to the virtual channels that fold into its
new channel, so
+ * watermarks/statuses can be aggregated across old channels merged on a
fan-in rescale. Without
+ * a rescale each group is a singleton and aggregation is verbatim
pass-through.
+ */
+ private static <T>
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>>
buildWatermarkMergeGroups(
+ Map<SubtaskConnectionDescriptor, VirtualChannel<T>>
gateVirtualChannels,
+ RescaleMappings channelMapping) {
+ RescaleMappings oldToNewMapping = channelMapping.invert();
+ Map<Integer, List<VirtualChannel<T>>> byNewChannel = new HashMap<>();
+ Map<SubtaskConnectionDescriptor, Integer> keyToNewChannel = new
HashMap<>();
+ gateVirtualChannels.forEach(
+ (key, vc) -> {
+ int newChannelIndex = newChannelIndexOf(key,
oldToNewMapping);
+ byNewChannel.computeIfAbsent(newChannelIndex, k -> new
ArrayList<>()).add(vc);
+ keyToNewChannel.put(key, newChannelIndex);
+ });
+
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>> mergeGroups
= new HashMap<>();
+ keyToNewChannel.forEach(
+ (key, newChannelIndex) -> mergeGroups.put(key,
byNewChannel.get(newChannelIndex)));
+ return mergeGroups;
Review Comment:
Can't we write this transformation as
```
gateVirtualChannels
.entrySet()
.map(e -> newEntryWithNewKey(e)) // using newChannelIndexOf
.collect(Collectors.groupingBy(MapEntry::getKey))
```
?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java:
##########
@@ -308,6 +371,41 @@ void filterAndRewrite(
}
}
+ /**
+ * Writes one element, aggregating non-records across the {@code
mergeGroup}: emit the group
+ * min watermark (suppressed until every merged channel has one), and
{@code ACTIVE} status
+ * while any merged channel is active. Records and latency markers
pass through verbatim.
+ */
+ private void emitAggregated(
+ StreamElement element,
+ List<VirtualChannel<T>> mergeGroup,
+ DataOutputSerializer outputSerializer)
+ throws IOException {
+ if (element.isWatermark()) {
+ Watermark minWatermark =
+ mergeGroup.stream()
+ .map(VirtualChannel::getLastWatermark)
+
.min(Comparator.comparingLong(Watermark::getTimestamp))
Review Comment:
Since we do this on the hot path, should we avoid:
1. O(N)
2. stream/conversion
?
The former might be premature optimisation, but the latter I think would be
an easy change
ditto: `anyActive` below
--
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]
