JunRuiLee commented on code in PR #26211:
URL: https://github.com/apache/flink/pull/26211#discussion_r1970803935
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphContext.java:
##########
@@ -76,12 +76,13 @@ public interface StreamGraphContext {
boolean modifyStreamNode(List<StreamNodeUpdateRequestInfo> requestInfos);
/**
- * Check whether all upstream nodes of the stream node have finished
executing.
+ * Check whether the upstream nodes of the stream node with pointed type
number have finished
+ * executing.
*
* @param streamNode the stream node that needs to be determined.
* @return true if all upstream nodes are finished, false otherwise.
*/
- boolean areAllUpstreamNodesFinished(ImmutableStreamNode streamNode);
+ boolean checkUpstreamNodesFinished(ImmutableStreamNode streamNode, Integer
typeNumber);
Review Comment:
@Nullable, and some additional comments are required to clarify the meaning
of a null value for typeNumber.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java:
##########
@@ -97,92 +102,113 @@ protected void tryOptimizeAdaptiveJoin(
adaptiveJoinNode, upstreamEdge.getTypeNumber(),
producedBytes);
}
- // If all upstream nodes have finished, we attempt to optimize the
AdaptiveJoin node.
- if (context.areAllUpstreamNodesFinished(adaptiveJoinNode)) {
- Long leftInputSize =
-
aggregatedInputBytesByTypeNumberAndNodeId.get(adaptiveJoinNode.getId()).get(1);
+ FlinkJoinType joinType = adaptiveJoin.getJoinType();
+ Long leftInputSize = null;
+ Long rightInputSize = null;
+
+ // When either input side of the adaptive join meets broadcast
requirements, the broadcast
+ // optimization can be immediately applied. This enables maximal
utilization of the
+ // {@link ForwardForUnspecifiedPartitioner} for unscheduled nodes on
the opposite side,
+ // effectively reducing network overhead.
Review Comment:
// When either input side of the adaptive join meets broadcast requirements,
the broadcast
// optimization can be immediately applied. This allows the opposite side to
produce data
// in a pointwise manner if it has not yet started scheduling, thereby
reducing the costs
// associated with shuffle writes and network overhead. Furthermore, in the
future,
// when multi-input support is available in the runtime, it could enable
chaining large input
// nodes with adaptive join nodes, further minimizing overall overhead.
--
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]
