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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
