rkhachatryan commented on code in PR #28459:
URL: https://github.com/apache/flink/pull/28459#discussion_r3532982436
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -473,11 +525,17 @@ private void createOperatorChainInfos(
final Map<Integer, OperatorChainInfo> chainEntryPoints =
buildAndGetChainEntryPoints(streamNodes,
jobVertexBuildContext);
+ chainEntryPoints.values().stream()
+ .filter(
+ chainInfo ->
+ streamGraph
+
.getStreamNode(chainInfo.getStartNodeId())
+ .getTransformationUID()
+ != null)
+ .forEach(chainInfo ->
generateHashesByStreamNodeId(chainInfo.getStartNodeId()));
Review Comment:
This eager `generateHashesByStreamNodeId` call inserts the UID-derived
hashes into the shared `hashes` map before the regular traversal. Since
`StreamGraphHasherV2#generateDeterministicHash` seeds every non-UID node hash
with `hashes.size()`, this shifts the positions of all non-UID nodes hashed
afterwards — so for mixed UID/non-UID graphs the adaptive path now produces
different `OperatorID`s/`JobVertexID`s than `StreamingJobGraphGenerator` for
the same program (and different from the adaptive path before this PR).
Example: sources A (no uid, node id 1) and B (uid, node id 2) — the legacy
hasher hashes A at position 0, but here B is hashed eagerly first, so A lands
at position 1 and gets a different hash. No existing test covers a mixed graph,
so `isJobGraphEquivalent` parity won't catch it.
Could we compute the sort key without mutating the map instead? For a UID
head the hash is by definition
`StreamGraphHasherV2.generateUserSpecifiedHash(uid)` (or simply sort by the UID
string — any deterministic total order works). That would remove this block and
its side effect entirely.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -447,6 +450,55 @@ private void
connectToFinishedUpStreamVertex(JobVertexBuildContext jobVertexBuil
}
}
+ private List<StreamEdge> getTransitiveInEdgesInOrder(
+ List<StreamEdge> transitiveInEdges, JobVertexBuildContext
jobVertexBuildContext) {
+ final List<StreamEdge> transitiveInEdgesInOrder =
+ transitiveInEdges.stream()
+ .sorted(
+ Comparator.comparing(
+ inEdge ->
getStartNodeId(inEdge.getSourceId())))
+ .collect(Collectors.toList());
+ final List<StreamEdge> uidTransitiveInEdges =
+ transitiveInEdgesInOrder.stream()
+ .filter(this::hasUidBackedUpstream)
+ .sorted(
+ Comparator.comparing(
+ inEdge ->
+ getStartNodeJobVertexId(
+ inEdge,
jobVertexBuildContext)))
+ .collect(Collectors.toList());
+
+ if (uidTransitiveInEdges.size() < 2) {
+ return transitiveInEdgesInOrder;
Review Comment:
Returning `transitiveInEdgesInOrder` here (the `uidTransitiveInEdges.size()
< 2` early return) drops the original insertion order (in-edge declaration
order) in favor of a sort by upstream start node id, even when no UID
reordering applies. For `b.union(a)` with `a` declared first, the gate order in
this reconnect path flips from `[b, a]` to `[a, b]`.
This actually aligns the adaptive path with the legacy generator (whose gate
order follows entry-point/node-id order), so it may be intentional — but it
contradicts the PR description's "Preserved legacy ordering for non-UID and
uidHash-only source heads" and isn't covered by a test. Could you either return
the unsorted list in this case, or state/test the parity change explicitly?
--
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]
