JunRuiLee commented on code in PR #25551:
URL: https://github.com/apache/flink/pull/25551#discussion_r1881896023
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java:
##########
@@ -151,16 +152,17 @@ private List<InputGateDeploymentDescriptor>
createInputGateDeploymentDescriptors
IntermediateDataSetID resultId =
consumedIntermediateResult.getId();
ResultPartitionType partitionType =
consumedIntermediateResult.getResultType();
- IndexRange subpartitionRange =
- executionVertex
- .getExecutionVertexInputInfo(resultId)
- .getSubpartitionIndexRange();
inputGates.add(
new InputGateDeploymentDescriptor(
resultId,
partitionType,
- subpartitionRange,
+ constructConsumedSubpartitionGroupByChannelRange(
+ executionVertex
Review Comment:
I'm confused about the mention of **channel** here. IMO, a channel is a
concept at the network layer, and during the deployment of a task in Flink, we
should not need to consider any channel-related concepts.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IndexRangeUtil.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Utils for {@link IndexRange}. */
+public class IndexRangeUtil {
+ public static List<IndexRange> mergeIndexRanges(Collection<IndexRange>
ranges) {
+ if (ranges == null || ranges.isEmpty()) {
+ return new ArrayList<>();
+ }
+
+ List<IndexRange> sortedRanges =
+ ranges.stream()
+
.sorted(Comparator.comparingInt(IndexRange::getStartIndex))
+ .collect(Collectors.toList());
+
+ List<IndexRange> merged = new ArrayList<>();
+ IndexRange current = sortedRanges.get(0);
+
+ for (int i = 1; i < ranges.size(); i++) {
+ IndexRange next = sortedRanges.get(i);
+ // <1,4>,<5,6>; <1,4>,<3,6>
Review Comment:
what's this comment meaning?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SsgNetworkMemoryCalculationUtils.java:
##########
@@ -190,15 +191,18 @@ static void getMaxInputChannelInfoForDynamicGraph(
IntermediateResultPartition resultPartition =
ejv.getGraph().getResultPartitionOrThrow((partitionGroup.getFirst()));
- IndexRange subpartitionIndexRange =
+
+ Map<IndexRange, IndexRange> consumedSubpartitionGroups =
vertex.getExecutionVertexInputInfo(
resultPartition.getIntermediateResult().getId())
- .getSubpartitionIndexRange();
+ .getConsumedSubpartitionGroups();
+
+ int inputChannelNums =
Review Comment:
I think this returned value does not represent the max channel number.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IndexRangeUtil.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Utils for {@link IndexRange}. */
+public class IndexRangeUtil {
+ public static List<IndexRange> mergeIndexRanges(Collection<IndexRange>
ranges) {
Review Comment:
javadoc
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java:
##########
@@ -406,4 +419,31 @@ public String toString() {
numLocalChannels, numRemoteChannels, numUnknownChannels);
}
}
+
+ public static int calculateNumInputChannels(
Review Comment:
Could you explain why the number of channels equals the size of the merged
index ranges instead of the size of the shuffle descriptors?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java:
##########
@@ -132,15 +149,18 @@ public ResultPartitionType getConsumedPartitionType() {
@Nonnegative
public int getConsumedSubpartitionIndex() {
Review Comment:
This method is never used, so we could remove it directly.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java:
##########
@@ -71,11 +72,11 @@ public class InputGateDeploymentDescriptor implements
Serializable {
private final ResultPartitionType consumedPartitionType;
/**
- * Range of the index of the consumed subpartition of each consumed
partition. This index
- * depends on the {@link DistributionPattern} and the subtask indices of
the producing and
+ * Range of the index of the consumed subpartition of each input channel
index ranges. This
Review Comment:
Could we just use the partition and subpartition concepts instead of
involving the confusing channel concept?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java:
##########
@@ -132,15 +149,18 @@ public ResultPartitionType getConsumedPartitionType() {
@Nonnegative
public int getConsumedSubpartitionIndex() {
+ checkState(consumedSubpartitionGroups.size() == 1);
+ IndexRange consumedSubpartitionIndexRange =
+ consumedSubpartitionGroups.values().iterator().next();
checkState(
consumedSubpartitionIndexRange.getStartIndex()
== consumedSubpartitionIndexRange.getEndIndex());
return consumedSubpartitionIndexRange.getStartIndex();
}
- /** Return the index range of the consumed subpartitions. */
- public IndexRange getConsumedSubpartitionIndexRange() {
- return consumedSubpartitionIndexRange;
+ /** Return the consumed subpartition groups by input channel range. */
+ public Map<IndexRange, IndexRange> getConsumedSubpartitionGroups() {
+ return consumedSubpartitionGroups;
Review Comment:
we can use the unmodifiedMap
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
