1996fanrui commented on code in PR #29064:
URL: https://github.com/apache/flink/pull/29064#discussion_r3928217233
##########
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:
Done — both switched to plain loops.
> O(N)
It's the recovery unspilling path and only runs per watermark/status (not
per record), so low risk. Let us kept O(N) as-is first.
##########
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:
Applied for the grouping, thanks.
It yields `Map<newChannelIndex, List<VC>>`, so a second pass is still needed
to re-key to old-channel key. Simplifies the front half but not the whole
method.
--
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]
