FrankYang0529 commented on code in PR #75190:
URL: https://github.com/apache/airflow/pull/75190#discussion_r4344179129
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -87,15 +87,31 @@ const TASK_MODULE = "airflow.sdk.coordinators.node";
*/
const TASK_LANGUAGE = "typescript";
-/** Python's `DAG.edge_info`, keying a group end by its join node as Python
does. */
-function serializeEdgeInfo(dag: Dag): SerializedValue {
+/**
+ * Python's `DAG.edge_info`, keying a group end by its join node as Python
does.
+ *
+ * This writes the labels in the order they were last set. Python writes a
label when it draws the
+ * edge. So when two labels land on one entry, the label set last is the one
kept.
+ */
+function serializeEdgeInfo(dag: Dag, graph: DagGraph): SerializedValue {
const groups = getDagTaskGroups(dag);
+ const grouped = new Set([...groups.values()].flatMap((group) =>
group.taskIds));
const info: Record<string, Record<string, SerializedValue>> = {};
- for (const { upstream, downstream, label } of getDagOrderEdges(dag)) {
+ const put = (from: string, to: string, label: string): void => {
+ (info[from] ??= {})[to] = { label };
+ };
+ for (const { upstream, downstream, label } of getDagLabelledEdges(dag)) {
if (label === undefined) continue;
const from = groups.has(upstream) ? upstream + DOWNSTREAM_JOIN_SUFFIX :
upstream;
const to = groups.has(downstream) ? downstream + UPSTREAM_JOIN_SUFFIX :
downstream;
- (info[from] ??= {})[to] = { label };
+ put(from, to, label);
+ // Python's `extract >> Label("x") >> group` also labels the edge from
extract to each root of
+ // the group. Like the Java SDK, this code labels the edges to the roots
for an edge drawn with
+ // `before` or `after`, but only when no group holds extract.
+ if (grouped.has(upstream)) continue;
Review Comment:
Yes, keeping `g1.e` as the end of the edge is deliberate: a label never
changes the ends of its edge. I expanded the comment to say what Python does
instead, and why this code labels no task edge when a group holds the upstream
task.
--
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]
