kaxil commented on code in PR #74230:
URL: https://github.com/apache/airflow/pull/74230#discussion_r4206219857
##########
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt:
##########
@@ -403,15 +519,67 @@ class BuilderProcessor : AbstractProcessor() {
}
/**
- * Rejects a task method whose wiring-view twin would clash with a member
- * the view or the wiring class already has: `depends`, `lit`, or a method
- * of `Object`.
+ * Rejects two tasks of the Dag sharing an ID, a task and a task group
+ * sharing one, and two task methods whose generated classes would collide.
+ */
+ private fun checkIds(scope: Scope) {
+ val declarations = scope.allTasks()
+ val taskIds = mutableSetOf<String>()
+ declarations.forEach { decl ->
+ require(taskIds.add(decl.id)) { "Tasks in Dag have duplicate ID:
${decl.id}" }
+ }
+ scope.allGroups().forEach { group ->
Review Comment:
This compares groups against tasks but not against each other, so two group
classes in one scope with `@Builder.TaskGroup(id = "checks")` compile fine and
only fail when the Dag is parsed, from `addGroup`'s duplicate check via
`Refs.createGroup`. Since duplicate task IDs are caught at compile time just
above, should duplicate group `fullId`s be too?
##########
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt:
##########
@@ -736,6 +933,13 @@ private val RESERVED_VIEW_NAMES =
"wait",
)
+/**
+ * What a group's view inherits from [Group] on top of [RESERVED_VIEW_NAMES].
+ * Only a group scope reserves them: a Dag's view extends [Deps], which has
+ * neither.
+ */
+private val RESERVED_GROUP_VIEW_NAMES = setOf("groupId", "nodes")
Review Comment:
This only reserves `groupId` and `nodes`, but a group view also inherits
`endpoints`, `before` and `after` from `Deps.Flow`. A task method `endpoints()`
(or a group class `Endpoints`) produces a javac error in the generated
interface instead of the clash message from `checkViewNames`. `before`/`after`
fail silently: with `@Builder.Task long before(long rows)` inside `Staging`,
`staging().before(extract())` resolves to the generated non-varargs
`before(Arg<? extends Long>)` over the inherited `before(Flow...)`, so it
registers the task `Staging.before` instead of ordering the group, and compiles
cleanly. Could the set include all three, ideally derived from
`Deps.TaskGroup`'s members so it can't drift? The KDoc here and at L550 still
links `[Group]`, which is now `Deps.TaskGroup`.
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -694,6 +694,66 @@ class that supplies only task bodies, for a Dag a Python
file declares, carries
for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's
position is what the
task receives. Wired inputs are the fallback, which is what a native Java
Dag always uses.
+Task groups
+~~~~~~~~~~~
+
+A task group gathers tasks that the Airflow UI shows as one node, as Python's
``TaskGroup`` does.
+Everything declared in a group carries the group's ID as a prefix, so task
``stage`` in group
+``staging`` is the task ``staging.stage``. On the interface surface,
``taskGroup`` declares a group on
+the Dag or inside another group, and the group declares its tasks:
+
+.. code-block:: java
+
+ var staging = dag.taskGroup("staging");
+ var stage = staging.task("stage", Stage.class); //
"staging.stage"
+ staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); //
"staging.checks.nulls"
+ extract.before(staging);
+
+With annotations, a ``@Builder.TaskGroup`` class holds the tasks of one group,
and nesting one in
+another nests the groups:
+
+.. code-block:: java
+
+ @Builder.TaskGroup // the group "Staging", after the
class
+ static class Staging {
+ @Builder.Task
+ public long stage(long rows) { ... } // the task "Staging.stage"
+
+ @Builder.TaskGroup(id = "checks")
+ static class Checks {
+ @Builder.Task
+ public void nulls(long staged) { ... } // "Staging.checks.nulls"
+ }
+ }
+
+ @Builder.Deps
+ static class Wiring implements EtlPipelineDeps {
Review Comment:
This wiring implements `EtlPipelineDeps`, but it drops the `transform`,
`load` and `audit` calls from the wiring above, and the page says an uncalled
task fails at Dag-parse time. Keeping those lines, or a `// ... wiring from
above` marker, would stop a copy-paste from failing.
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -694,6 +694,66 @@ class that supplies only task bodies, for a Dag a Python
file declares, carries
for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's
position is what the
task receives. Wired inputs are the fallback, which is what a native Java
Dag always uses.
+Task groups
+~~~~~~~~~~~
+
+A task group gathers tasks that the Airflow UI shows as one node, as Python's
``TaskGroup`` does.
+Everything declared in a group carries the group's ID as a prefix, so task
``stage`` in group
+``staging`` is the task ``staging.stage``. On the interface surface,
``taskGroup`` declares a group on
+the Dag or inside another group, and the group declares its tasks:
+
+.. code-block:: java
+
+ var staging = dag.taskGroup("staging");
+ var stage = staging.task("stage", Stage.class); //
"staging.stage"
+ staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); //
"staging.checks.nulls"
+ extract.before(staging);
+
+With annotations, a ``@Builder.TaskGroup`` class holds the tasks of one group,
and nesting one in
+another nests the groups:
+
+.. code-block:: java
+
+ @Builder.TaskGroup // the group "Staging", after the
class
+ static class Staging {
+ @Builder.Task
+ public long stage(long rows) { ... } // the task "Staging.stage"
+
+ @Builder.TaskGroup(id = "checks")
+ static class Checks {
+ @Builder.Task
+ public void nulls(long staged) { ... } // "Staging.checks.nulls"
+ }
+ }
+
+ @Builder.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ var rows = extract();
+ var staged = staging().stage(rows);
+ staging().checks().nulls(staged);
+ extract().before(staging());
+ }
+ }
+
+The generated view nests the same way, so a group is both the namespace of
what it holds and a point
+in the flow: ``staging().checks().nulls(staged)`` reaches a task, and
``extract().before(staging())``
+orders the whole group. Task method names scope to their own group, so two
groups can each declare
+``run()``. A group class is ``static``, non-private, and needs a no-argument
constructor, because the
+generated task bodies instantiate it.
+
+A group stands at either end of ``before``, ``after`` and ``Flow.of``. As an
upstream it stands for
+its leaves, the tasks nothing else in the group runs after; as a downstream,
for its roots, the tasks
+that run after nothing else in the group. A group ID contains only ASCII
letters, digits,
+underscores, or dashes, and no task or other group in the Dag can share it.
+
+.. note::
+
+ A group's endpoints are read once, from everything the group holds by the
time the Dag is
Review Comment:
Only group membership is read at the end. Group edges are still applied in
the order they were drawn, which `expandGroupEdges`'s KDoc says and
`shouldResolveGroupEndpointsAgainstEarlierEdges` pins. With `outer` holding `t`
and nested `inner` holding `i`, `inner.before(t); extract.before(outer);` gives
`outer.t` the upstreams {`outer.inner.i`}, but the swapped order gives
{`extract`, `outer.inner.i`}, so under `one_success` `t` can start before
`inner` has run. Could this note and the matching paragraph in `TaskGroupRef`'s
KDoc say that membership is final but group edges resolve in drawing order? A
test for the swapped order would pin it.
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TaskGroupTest.kt:
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+
+internal class TaskGroupTest {
+ private fun upstreamIds(
+ dag: DagDef,
+ taskId: String,
+ ) = dag.expandGroupEdges().upstreamsOf(dag.tasks.getValue(taskId))
+
+ @Test
+ @DisplayName("Should prefix the IDs of tasks and groups declared in a group")
+ fun shouldPrefixIdsDeclaredInGroup() {
+ val dag = DagDef("d")
+ val staging = dag.taskGroup("staging")
+ val stage = staging.task<Unit>("stage", NoopTask::class.java)
+ val checks = staging.taskGroup("checks")
+ checks.task<Unit>("nulls", NoopTask::class.java)
+
+ assertEquals("staging.stage", stage.def.id)
+ assertEquals("staging.checks", checks.id)
+ assertEquals(listOf("staging.stage", "staging.checks.nulls"),
dag.tasks.keys.toList())
+ assertEquals(listOf("staging", "staging.checks"), dag.groups.keys.toList())
+ }
+
+ @Test
+ @DisplayName("Should reject a group ID that is not a plain identifier")
+ fun shouldRejectInvalidGroupId() {
+ val error = assertThrows(IllegalArgumentException::class.java) {
DagDef("d").taskGroup("a.b") }
+
+ assertEquals(
+ "Task group ID 'a.b' must contain only ASCII letters, digits,
underscores, or dashes",
+ error.message,
+ )
+ }
+
+ @Test
+ @DisplayName("Should reject a group ID that a task already uses")
+ fun shouldRejectGroupIdTakenByTask() {
+ val dag = DagDef("d")
+ dag.task<Unit>("staging", NoopTask::class.java)
+
+ val error = assertThrows(IllegalArgumentException::class.java) {
dag.taskGroup("staging") }
+
+ assertEquals("Dag 'd' already has a task or task group with ID: staging",
error.message)
+ }
+
+ @Test
+ @DisplayName("Should reject a task ID that a group already uses")
+ fun shouldRejectTaskIdTakenByGroup() {
+ val dag = DagDef("d")
+ dag.taskGroup("staging")
+
+ val error =
+ assertThrows(IllegalArgumentException::class.java) {
dag.task<Unit>("staging", NoopTask::class.java) }
+
+ assertEquals("Dag 'd' already has a task group with ID: staging",
error.message)
+ }
+
+ @Test
+ @DisplayName("Should wire a group upstream from its leaves and downstream to
its roots on registration")
+ fun shouldExpandGroupEdgesOntoRootsAndLeaves() {
+ val dag = DagDef("d")
+ val extract = dag.task<Unit>("extract", NoopTask::class.java)
+ val staging = dag.taskGroup("staging")
+ val stage = staging.task<Unit>("stage", NoopTask::class.java)
+ staging.taskGroup("checks").task<Unit>("nulls",
NoopTask::class.java).after(stage)
+ val publish = dag.taskGroup("publish")
+ publish.task<Unit>("push", NoopTask::class.java)
+ val load = dag.task<Unit>("load", NoopTask::class.java)
+ extract.before(staging)
+ staging.before(publish)
+ publish.before(load)
+
+ Bundle().register(dag)
+
+ assertEquals(setOf("extract"), upstreamIds(dag, "staging.stage"))
+ assertEquals(setOf("staging.stage"), upstreamIds(dag,
"staging.checks.nulls"))
+ assertEquals(setOf("staging.checks.nulls"), upstreamIds(dag,
"publish.push"))
+ assertEquals(setOf("publish.push"), upstreamIds(dag, "load"))
+ }
+
+ @Test
+ @DisplayName("Should record each group's own edges the way Python's
TaskGroup does")
+ fun shouldRecordGroupEdgesOnGroups() {
+ val dag = DagDef("d")
+ val extract = dag.task<Unit>("extract", NoopTask::class.java)
+ val staging = dag.taskGroup("staging")
+ staging.task<Unit>("stage", NoopTask::class.java)
+ val publish = dag.taskGroup("publish")
+ publish.task<Unit>("push", NoopTask::class.java)
+ val load = dag.task<Unit>("load", NoopTask::class.java)
+ extract.before(staging)
+ staging.before(publish)
+ publish.before(load)
+
+ Bundle().register(dag)
+
+ val edges = dag.expandGroupEdges()
+ assertEquals(setOf("extract"), edges.edgesOf(staging.id).upstreamTaskIds)
+ assertEquals(setOf("publish"),
edges.edgesOf(staging.id).downstreamGroupIds)
+ assertEquals(emptySet<String>(),
edges.edgesOf(staging.id).downstreamTaskIds)
+ assertEquals(setOf("staging"), edges.edgesOf(publish.id).upstreamGroupIds)
+ assertEquals(setOf("staging.stage"),
edges.edgesOf(publish.id).upstreamTaskIds)
+ assertEquals(setOf("load"), edges.edgesOf(publish.id).downstreamTaskIds)
+ }
+
+ @Test
+ @DisplayName("Should step over a group with no tasks to the tasks beyond it")
+ fun shouldStepOverEmptyGroup() {
+ val dag = DagDef("d")
+ val extract = dag.task<Unit>("extract", NoopTask::class.java)
+ val empty = dag.taskGroup("empty")
+ val load = dag.task<Unit>("load", NoopTask::class.java)
+ extract.before(empty)
+ empty.before(load)
+
+ Bundle().register(dag)
+
+ assertEquals(setOf("extract"), upstreamIds(dag, "load"))
+ }
+
+ @Test
+ @DisplayName("Should step over an empty nested group to the tasks of the
group holding it")
+ fun shouldStepOverEmptyNestedGroup() {
Review Comment:
Neither this test nor `shouldStepOverEmptyGroup` can tell the new fallback
order (leaves, then `upstreamTaskIds`, then the parent) from parent-first: here
`inner` has no incoming edge, and the other has no parent. Adding
`extract.before(inner)` before `inner.before(load)` and asserting `load` waits
on {`extract`} rather than {`outer.t`} would pin it.
--
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]
