You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2026/10/20 04:28:10 UTC
[PR] Go SDK: Write the edge label text before the task [airflow]
FrankYang0529 opened a new pull request, #75185:
URL: https://github.com/apache/airflow/pull/75185
Change label form from `airflow.Label(node, text)` to `airflow.Label(text).On(node)` as Java SDK's edge labels (#74523).
- `airflow.Label(text)` returns a `*airflow.LabelRef`. `(*LabelRef).On(node, more...)` returns the `Node` to pass to `Before` or `After`. The signature of `On` requires at least one node, so the compiler rejects a call with none. When `On` gets several nodes, the edges to all of them get the same label, as in Python's `a >> Label("forks") >> [b, c]`.
- `airflow.Label` now panics on text that is only white space, not only on empty text. `Before` and `After` now panic when their receiver carries a label.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes - Claude Code
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #75185:
URL: https://github.com/apache/airflow/pull/75185#discussion_r4338596896
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
+ const where = "airflow.LabelRef.On"
+ if l == nil || l.text == "" {
+ panic(where + ": airflow.Label did not return the *airflow.LabelRef")
}
- labelling := endpoints("airflow.Label: node", node)
- labelled := make(nodeSet, len(labelling))
- for i, endpoint := range labelling {
- labelled[i] = nodeEndpoint{task: endpoint.task, group: endpoint.group, label: text}
+ nodes := append([]Node{node}, more...)
+ labelled := make(nodeSet, 0, len(nodes))
+ for i, labelling := range nodes {
+ name := "node"
+ if i > 0 {
+ name = fmt.Sprintf("more[%d]", i-1)
Review Comment:
Passing the index as you suggested. Thanks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #75185:
URL: https://github.com/apache/airflow/pull/75185#discussion_r4338595676
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
Review Comment:
For `Label("x").On(a).Before(b)` to fail to compile, `On` would have to return a type without `Before` and `After`. The value from `On` still has to go into `loaded.Before(...)`, so `Before` could no longer take `...Node`. `Before` would need a new argument type, say `Arg`, that both a `Node` and the labelled value implement. Code that spreads a slice of nodes would then have to build a `[]airflow.Arg`, because Go does not convert a `[]airflow.Node` into a `[]airflow.Arg`:
```go
var shards []airflow.Node
for _, region := range regions {
shards = append(shards, dag.Task(count, airflow.TaskSpec{TaskID: region}))
}
loaded.Before(shards...) // compiles today, but not if Before takes ...airflow.Arg
```
I prefer to keep the run-time check on the receiver.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #75185:
URL: https://github.com/apache/airflow/pull/75185#discussion_r4338595381
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
+ const where = "airflow.LabelRef.On"
+ if l == nil || l.text == "" {
+ panic(where + ": airflow.Label did not return the *airflow.LabelRef")
}
- labelling := endpoints("airflow.Label: node", node)
- labelled := make(nodeSet, len(labelling))
- for i, endpoint := range labelling {
- labelled[i] = nodeEndpoint{task: endpoint.task, group: endpoint.group, label: text}
+ nodes := append([]Node{node}, more...)
+ labelled := make(nodeSet, 0, len(nodes))
+ for i, labelling := range nodes {
+ name := "node"
+ if i > 0 {
+ name = fmt.Sprintf("more[%d]", i-1)
+ }
+ if labelling == nil {
+ panic(fmt.Sprintf("%s: %s is a nil airflow.Node", where, name))
+ }
+ for _, endpoint := range endpoints(where+": "+name, labelling) {
+ labelled = append(
+ labelled, nodeEndpoint{task: endpoint.task, group: endpoint.group, label: l.text},
Review Comment:
- `On` now returns its own type, `labelledSet`, instead of the `nodeSet` that `Before` and `After` return. So `checkReceiverUnlabelled` catches `Label("x").On(a).Before(b)` by checking the type of the receiver, instead of looking at each endpoint.
- `Before` and `After` still return the nodes they were given without the labels, so that a label does not carry over when the result goes into the next verb.
##########
go-sdk/airflow/node.go:
##########
@@ -347,6 +393,25 @@ func declareEdges(receiver Node, nodes []Node, dir edgeDir) Node {
return args.unlabelled()
}
+// checkReceiverUnlabelled panics if the receiver of an edge verb carries a label. Only the Node
Review Comment:
Added test case to `TestAVerbWithAnEmptyReceiverRejectsALabel`.
##########
go-sdk/airflow/node_test.go:
##########
@@ -184,18 +184,36 @@ func TestLabelLabelsEachEdgeOfAFanOut(t *testing.T) {
processed := orderedTask(t, dag, "process")
emptyNotice := orderedTask(t, dag, "notify_empty")
- checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+ checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
assertEdgeLabel(t, dag, "check", "process", "rows found")
assertEdgeLabel(t, dag, "check", "notify_empty", "no rows")
}
+// TestLabelOnLabelsTheEdgeToEachNodeItIsGiven checks the Go form of Python's
+// loaded >> Label("forks") >> [notify_empty, staging, notify_suspicious].
+func TestLabelOnLabelsTheEdgeToEachNodeItIsGiven(t *testing.T) {
Review Comment:
Added it. Thanks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #75185:
URL: https://github.com/apache/airflow/pull/75185
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #75185:
URL: https://github.com/apache/airflow/pull/75185#issuecomment-6330450545
### Backport successfully created: v3-4-test
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-4-test</td>
<td><a href="https://github.com/apache/airflow/pull/75365"><img src="https://img.shields.io/badge/PR-75365-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #75185:
URL: https://github.com/apache/airflow/pull/75185#discussion_r4334454924
##########
go-sdk/airflow/node.go:
##########
@@ -347,6 +393,25 @@ func declareEdges(receiver Node, nodes []Node, dir edgeDir) Node {
return args.unlabelled()
}
+// checkReceiverUnlabelled panics if the receiver of an edge verb carries a label. Only the Node
Review Comment:
The following case still drops the label silently:
```go
var none []Node
loaded.Before(none...).Before(Label("always").On(cleaned)) // "always" is lost, no error
```
It returns at the `len(ends) == 0` exit in `declareEdges` (line 353), and `TestAVerbThatDeclaresNoEdgeReturnsNoLabel` pins that behavior. Let's panic there too.
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
Review Comment:
`On` names the offending argument for a nil `Node` and a nil `*TaskGroupRef` (`more[1] is a nil *airflow.TaskGroupRef`), but a nil `*TaskRef` falls through to the verb, which panics with `airflow.Node.Before: got a nil *airflow.TaskRef` and no position. With several nodes the author cannot tell which one it was. Worth checking it here so the message keeps the position.
##########
go-sdk/airflow/node_test.go:
##########
@@ -184,18 +184,36 @@ func TestLabelLabelsEachEdgeOfAFanOut(t *testing.T) {
processed := orderedTask(t, dag, "process")
emptyNotice := orderedTask(t, dag, "notify_empty")
- checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+ checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
assertEdgeLabel(t, dag, "check", "process", "rows found")
assertEdgeLabel(t, dag, "check", "notify_empty", "no rows")
}
+// TestLabelOnLabelsTheEdgeToEachNodeItIsGiven checks the Go form of Python's
+// loaded >> Label("forks") >> [notify_empty, staging, notify_suspicious].
+func TestLabelOnLabelsTheEdgeToEachNodeItIsGiven(t *testing.T) {
Review Comment:
This is the only test of `On` with a group, and it stops at the group edge. Without `Bundle().Register(dag)` it never covers the exception the ADR documents, so a regression that drops the label on the expanded `load -> staging.stage` edge would still pass. Worth adding:
```go
Bundle().Register(dag)
assertEdgeLabel(t, dag, "load", "staging.stage", "forks")
```
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
Review Comment:
Altitude: `On` returns a plain `Node`, so `Label("x").On(a).Before(b)` compiles and fails only when the Dag file is parsed, and every `Before`/`After` call pays for `checkReceiverUnlabelled` to catch it. A distinct labelled type that the verbs accept as an argument but that has no `Before`/`After` methods would reject the misuse at compile time and let the runtime check go.
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
+ const where = "airflow.LabelRef.On"
+ if l == nil || l.text == "" {
+ panic(where + ": airflow.Label did not return the *airflow.LabelRef")
}
- labelling := endpoints("airflow.Label: node", node)
- labelled := make(nodeSet, len(labelling))
- for i, endpoint := range labelling {
- labelled[i] = nodeEndpoint{task: endpoint.task, group: endpoint.group, label: text}
+ nodes := append([]Node{node}, more...)
+ labelled := make(nodeSet, 0, len(nodes))
+ for i, labelling := range nodes {
+ name := "node"
+ if i > 0 {
+ name = fmt.Sprintf("more[%d]", i-1)
+ }
+ if labelling == nil {
+ panic(fmt.Sprintf("%s: %s is a nil airflow.Node", where, name))
+ }
+ for _, endpoint := range endpoints(where+": "+name, labelling) {
+ labelled = append(
+ labelled, nodeEndpoint{task: endpoint.task, group: endpoint.group, label: l.text},
Review Comment:
Simplification: the label is copied onto every `nodeEndpoint`, then every verb allocates a fresh set through `unlabelled()` to strip it, and `checkReceiverUnlabelled` scans each endpoint for a non-empty one. A wrapper holding the text and the set once, say `labelledSet{nodes nodeSet; text string}` that `endpoints` expands for arguments, would turn the receiver check into a single type assertion and drop the per-call copy.
##########
go-sdk/adr/0008-native-dag-interface.md:
##########
@@ -196,8 +198,8 @@ const (
- **A group with no task is stepped over.** An edge to it continues along each edge from it, and an edge from it back along each edge to it, whenever those were declared, as the TypeScript SDK does. So `extracted.Before(empty).Before(loaded)` runs `load` after `extract`, as Python's `extract >> empty >> load` does.
What Python does with an empty group depends on how and when its edges are declared: `load << empty << extract` adds no edge, and an empty group with no task before it falls back to the last tasks of the enclosing group or of the whole Dag, which can make a task depend on itself.
- **An edge cannot connect a group to what it holds.** Python's `group >> node_in_group` orders the last tasks of the group before the first tasks of a node inside it, which fails as a cycle whenever that node holds a task. The edge verb rejects the edge when it is declared.
-- **A label stays on the edge it is declared on.** A label on a group edge labels none of the task edges that the group edge stands for, and a labelled edge keeps its ends.
- Python's behavior depends on which end is the receiver and which groups hold the ends. When no group holds `extract`, `extract >> Label("rows") >> group` labels each of those task edges as well. When `a` and `b` are in different groups, `a >> Label("x") >> b` replaces `a` with its group, so every last task of `a`'s group runs before `b`. The Go SDK does neither.
+- **A label stays on the edge it is declared on, with one exception.** A label on a group edge labels that group edge. A group edge from a task in no group to a group with tasks also puts its label on the task edges from that task to the first tasks of the group. Python's `extract >> Label("rows") >> group` does the same. If one of those task edges was also declared on its own, or an earlier group edge added it, that task edge keeps the label it has. If it has no label, it stays unlabelled. Python labels those task edges when `>>` runs, so in Python the label written last wins.
+ A label does not change the ends of its edge. When `a` and `b` are in different groups, Python's `a >> Label("x") >> b` replaces `a` with its group, so all the last tasks of `a`'s group run before `b`. The Go SDK keeps the edge between `a` and `b`.
Review Comment:
Consequences lists every other divergence from Python (enum rejection, empty groups, label ends), but not the new one in this PR: `Before` and `After` panic on a labelled receiver, where Python's `Label("x") >> b` drops the label without an error.
```suggestion
A label does not change the ends of its edge. When `a` and `b` are in different groups, Python's `a >> Label("x") >> b` replaces `a` with its group, so all the last tasks of `a`'s group run before `b`. The Go SDK keeps the edge between `a` and `b`.
- **A label on the receiver of an edge verb is rejected.** `Before` and `After` panic, because that label has no edge to go on. Python's `Label("x") >> b` drops it without an error.
```
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
+ const where = "airflow.LabelRef.On"
+ if l == nil || l.text == "" {
+ panic(where + ": airflow.Label did not return the *airflow.LabelRef")
}
- labelling := endpoints("airflow.Label: node", node)
- labelled := make(nodeSet, len(labelling))
- for i, endpoint := range labelling {
- labelled[i] = nodeEndpoint{task: endpoint.task, group: endpoint.group, label: text}
+ nodes := append([]Node{node}, more...)
+ labelled := make(nodeSet, 0, len(nodes))
+ for i, labelling := range nodes {
+ name := "node"
+ if i > 0 {
+ name = fmt.Sprintf("more[%d]", i-1)
Review Comment:
Minor: the `more[%d]` string and the `where+": "+name` concatenation are built for every node on every call, but are only read if `On` panics. Formatting them inside the panic branches, or passing the index to `endpoints`, would keep the normal path allocation free.
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
Review Comment:
This still says a group label stays on the group edge, but `TaskGroupRef.Before` (task_group.go:185-188) and the ADR bullet rewritten in this PR both say it also reaches the task edges into the group's first tasks. The public godoc and the ADR now disagree.
```suggestion
// A label on an edge to or from a task group labels that group edge. On an edge from a task that
// no group holds to a group with tasks, it also labels the edges to the first tasks of the group,
// as [TaskGroupRef.Before] describes. A label on an edge between two tasks stays on that edge
// too, even when the tasks are in different task groups. In that case Python can replace the
// receiver of the verb with a group that holds it, which makes the edge connect other tasks.
```
##########
go-sdk/airflow/node.go:
##########
@@ -178,46 +179,90 @@ func (t *TaskRef) Before(nodes ...Node) Node { return declareEdges(t, nodes, dir
// cleanup << extract << start.
func (t *TaskRef) After(nodes ...Node) Node { return declareEdges(t, nodes, dirAfter) }
-// Label puts text on the edge that an edge verb declares to node, as Python's
-// loaded >> Label("when empty") >> notify_empty does:
+// LabelRef is an edge label made by [Label]. [LabelRef.On] returns a Node. To label edges, pass
+// that Node to Before or After.
+type LabelRef struct {
+ text string
+}
+
+// Label returns a label with the given text. The Python code
+// loaded >> Label("when empty") >> notify_empty becomes:
+//
+// loaded.Before(Label("when empty").On(emptyNotice))
+//
+// The text comes before the node passed to On, so it sits between the two tasks in the code. The
+// Airflow UI draws the label between the same two tasks in the graph.
+//
+// Label panics if text is empty or holds only white space.
+func Label(text string) *LabelRef {
+ if strings.TrimSpace(text) == "" {
+ panic("airflow.Label: got a blank label; pass the text to put on the edge")
+ }
+ return &LabelRef{text: text}
+}
+
+// On returns node, and each node in more, as one Node. Pass that Node to an edge verb, and the verb
+// puts the label on each edge between its receiver and one of those nodes. With After, those edges
+// point at the receiver:
+//
+// cleaned.After(Label("always").On(loaded)) // loaded >> Label("always") >> cleanup
//
-// loaded.Before(Label(emptyNotice, "when empty"))
+// Each Node passed to Before or After can carry its own label, so the edges that one call declares
+// can have different labels:
//
-// The label wraps the endpoint rather than the call, so each edge of a fan-out can carry a label
-// of its own:
+// checked.Before(Label("rows found").On(processed), Label("no rows").On(emptyNotice))
//
-// checked.Before(Label(processed, "rows found"), Label(emptyNotice, "no rows"))
+// When On is given more than one node, the edge to or from each of those nodes gets the same
+// label. Python's loaded >> Label("forks") >> [notify_empty, notify_suspicious] becomes:
//
-// A label is the Node's only in the verb it is passed to. The Node that Label returns stands for
-// node itself, and the Node a verb returns carries no label on either side of the next verb. So a
-// label on the receiver of a verb has no edge to land on and is dropped, the way Python's
-// Label("x") >> b alone sets no label.
+// loaded.Before(Label("forks").On(emptyNotice, suspiciousNotice))
+//
+// A label is the Node's only in the verb it is passed to. The Node that On returns stands for the
+// nodes themselves. The Node that a verb returns carries no label into the next verb, either as
+// its receiver or as an argument. Before and After panic when their receiver carries a label,
+// because that label has no edge to go on. Pass the Node that On returns only as an argument.
+// Python's Label("x") >> b drops the label without an error instead. When labels nest, the outer
+// label is used: loaded.Before(Label("outer").On(Label("inner").On(cleaned))) labels the edge
+// "outer".
//
// A label declared on an edge that already carries one replaces it, as Python's DAG.set_edge_info
// does.
//
// An [Inputs] edge is labelled by declaring it again, which is idempotent:
//
// transformed := dag.Task(transform, Inputs(extracted))
-// extracted.Before(Label(transformed, "rows"))
+// extracted.Before(Label("rows").On(transformed))
//
// A label on an edge to or from a task group stays on that edge, as [TaskGroupRef.Before]
// describes. A label on an edge between two tasks stays on that edge too, even when the tasks are
// in different task groups. In that case Python can replace the receiver of the verb with a group
// that holds it, which makes the edge connect other tasks.
//
-// Label panics if node is nil or text is empty.
-func Label(node Node, text string) Node {
- if node == nil {
- panic("airflow.Label: got a nil airflow.Node")
- }
- if text == "" {
- panic("airflow.Label: got an empty label; pass the text to put on the edge")
+// On panics if:
+// - l is nil or points to a zero LabelRef
+// - node or a node in more is nil, is a nil *TaskGroupRef, or is a struct that embeds a Node
+//
+// On does not panic on a nil *TaskRef. The edge verb that takes the Node panics on it instead.
+func (l *LabelRef) On(node Node, more ...Node) Node {
+ const where = "airflow.LabelRef.On"
+ if l == nil || l.text == "" {
+ panic(where + ": airflow.Label did not return the *airflow.LabelRef")
}
- labelling := endpoints("airflow.Label: node", node)
- labelled := make(nodeSet, len(labelling))
- for i, endpoint := range labelling {
- labelled[i] = nodeEndpoint{task: endpoint.task, group: endpoint.group, label: text}
+ nodes := append([]Node{node}, more...)
+ labelled := make(nodeSet, 0, len(nodes))
+ for i, labelling := range nodes {
Review Comment:
Reuse: this loop repeats the argument flattening in `declareEdges` (lines 316-321), with a different naming scheme (`node`/`more[i]` here, `nodes[i]` there). A shared `flattenNodes(where, names, nodes)` helper would keep the node validation in one place so the two cannot drift.
##########
go-sdk/adr/0008-native-dag-interface.md:
##########
@@ -153,8 +153,10 @@ type Node interface {
node()
}
-// Label carries an edge label into Before or After. The Node it returns stands for node itself.
-func Label(node Node, text string) Node
+// Label makes an edge label, and On carries the label into Before or After. The Node that On
+// returns stands for node and each node in more.
+func Label(text string) *LabelRef
Review Comment:
The block references `*LabelRef` without declaring the type.
```suggestion
type LabelRef struct{ /* unexported */ }
func Label(text string) *LabelRef
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Go SDK: Write the edge label text before the task [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #75185:
URL: https://github.com/apache/airflow/pull/75185#discussion_r4338594296
##########
go-sdk/adr/0008-native-dag-interface.md:
##########
@@ -153,8 +153,10 @@ type Node interface {
node()
}
-// Label carries an edge label into Before or After. The Node it returns stands for node itself.
-func Label(node Node, text string) Node
+// Label makes an edge label, and On carries the label into Before or After. The Node that On
+// returns stands for node and each node in more.
+func Label(text string) *LabelRef
Review Comment:
Added `LabelRef`. Thanks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org