jason810496 commented on code in PR #74438:
URL: https://github.com/apache/airflow/pull/74438#discussion_r4218526087
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt:
##########
@@ -271,9 +337,84 @@ class TaskTest {
): Unit = throw UnsupportedOperationException("not used in test")
override fun clearTaskStateStore(tiId: UUID): Unit = throw
UnsupportedOperationException("not used in test")
+
+ override fun skipDownstreamTasks(taskIds: List<String>): Unit = throw
UnsupportedOperationException("not used in test")
},
)
+ /** Decides what [decision] holds; a null one throws, standing for a
condition body that fails. */
+ class TestCondition : ConditionTask {
+ override fun decide(
+ context: Context,
+ client: Client,
+ ): Boolean = decision ?: throw IllegalStateException("boom")
+
+ companion object {
+ var decision: Boolean? = true
+ }
+ }
+
+ /** Records what a deciding task pushed and asked to skip, in order. */
Review Comment:
Done in f118e9e22d8: the fake records one ordered event list the assertions
walk, and CommTest now covers the skipDownstreamTasks round trip and its
empty-list short circuit.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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 kotlin.Throws
+
+/**
+ * A task whose boolean decides which of two tasks runs; the other is skipped.
+ *
+ * Register one with [DagDef.If], then name the sides with [ConditionRef.then]
+ * and [ConditionRef.orElse]:
+ *
+ * ```java
+ * public class HasRows implements ConditionTask {
+ * @Override
+ * public boolean decide(Context context, Client client) {
+ * return ((Number) client.getXCom("extract")).longValue() > 0;
+ * }
+ * }
+ * ```
+ *
+ * The SDK runs [decide] and pushes its result as this task's return value, so
+ * [execute] is never called.
+ *
+ * @see DagDef.If
+ */
+interface ConditionTask : Task {
+ /**
+ * Decides which side of the condition runs.
+ *
+ * Any exception thrown marks the task instance as failed, and nothing is
+ * then pushed or skipped.
+ *
+ * @param context Runtime context for the current execution workload.
+ * @param client Client for Airflow API calls scoped to this execution.
+ * @return True to run the [ConditionRef.then] side, false the
+ * [ConditionRef.orElse] side.
+ * @throws Exception on failure; the task instance is marked failed.
+ */
+ @Throws(Exception::class)
+ fun decide(
+ context: Context,
+ client: Client,
+ ): Boolean
+
+ /** Never called: the SDK runs a condition through [decide]. */
+ override fun execute(
Review Comment:
Done in 41ac9d3ca0d: Bundle.register now requires a task's class to decide
exactly when the task has a decider, so this fails at import.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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 kotlin.Throws
+
+/**
+ * A task whose boolean decides which of two tasks runs; the other is skipped.
+ *
+ * Register one with [DagDef.If], then name the sides with [ConditionRef.then]
+ * and [ConditionRef.orElse]:
+ *
+ * ```java
+ * public class HasRows implements ConditionTask {
+ * @Override
+ * public boolean decide(Context context, Client client) {
+ * return ((Number) client.getXCom("extract")).longValue() > 0;
+ * }
+ * }
+ * ```
+ *
+ * The SDK runs [decide] and pushes its result as this task's return value, so
+ * [execute] is never called.
+ *
+ * @see DagDef.If
+ */
+interface ConditionTask : Task {
+ /**
+ * Decides which side of the condition runs.
+ *
+ * Any exception thrown marks the task instance as failed, and nothing is
+ * then pushed or skipped.
+ *
+ * @param context Runtime context for the current execution workload.
+ * @param client Client for Airflow API calls scoped to this execution.
+ * @return True to run the [ConditionRef.then] side, false the
+ * [ConditionRef.orElse] side.
+ * @throws Exception on failure; the task instance is marked failed.
+ */
+ @Throws(Exception::class)
+ fun decide(
+ context: Context,
+ client: Client,
+ ): Boolean
+
+ /** Never called: the SDK runs a condition through [decide]. */
+ override fun execute(
+ context: Context,
+ client: Client,
+ ): Unit =
+ throw IllegalStateException(
+ "Condition '${javaClass.name}' runs through decide(), so execute() is
never called",
+ )
+}
+
+/**
+ * A condition registered with a Dag: name the task each outcome runs.
+ *
+ * ```java
+ * dag.If(HasRows.class).then(load).orElse(reportEmpty);
+ * ```
+ *
+ * A named task runs after the condition, so naming it records that edge, as
+ * [Deps.Flow.before] would. The condition skips only the side not taken; what
+ * runs after both sides needs a trigger rule that tolerates one skipped
+ * upstream, such as `none_failed_min_one_success`.
+ *
+ * @see DagDef.If
+ */
+class ConditionRef private constructor(
+ private val ref: TaskRef<Boolean>,
+ private val sides: ConditionDef,
+) : Deps.Flow {
+ companion object {
+ /**
+ * @suppress
+ *
+ * Marks a registered task as a condition. Public so a generated wiring
+ * view can call it; user code reaches a condition through [DagDef.If].
+ */
+ @JvmStatic
+ fun of(ref: TaskRef<Boolean>): ConditionRef {
+ val sides = ConditionDef()
+ require(ref.def.decider == null) {
Review Comment:
Done in ec0c071cd3f: of reuses the existing ConditionDef, so a no-arg view
method can name the sides across statements, covered by a BuilderTest and a
RefsTest.
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ConditionTest.kt:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.apache.airflow.sdk.execution.serializeDag
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+
+/** A condition the tests register; what it decides is driven from the runtime
tests. */
+class HasRows : ConditionTask {
+ override fun decide(
+ context: Context,
+ client: Client,
+ ): Boolean = decision
+
+ companion object {
+ var decision: Boolean = true
Review Comment:
Done in dbe29e41a1f: the decider returns true directly, and the field
nothing asserted is gone.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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 kotlin.Throws
+
+/**
+ * A task whose boolean decides which of two tasks runs; the other is skipped.
+ *
+ * Register one with [DagDef.If], then name the sides with [ConditionRef.then]
+ * and [ConditionRef.orElse]:
+ *
+ * ```java
+ * public class HasRows implements ConditionTask {
+ * @Override
+ * public boolean decide(Context context, Client client) {
+ * return ((Number) client.getXCom("extract")).longValue() > 0;
+ * }
+ * }
+ * ```
+ *
+ * The SDK runs [decide] and pushes its result as this task's return value, so
+ * [execute] is never called.
+ *
+ * @see DagDef.If
+ */
+interface ConditionTask : Task {
+ /**
+ * Decides which side of the condition runs.
+ *
+ * Any exception thrown marks the task instance as failed, and nothing is
+ * then pushed or skipped.
+ *
+ * @param context Runtime context for the current execution workload.
+ * @param client Client for Airflow API calls scoped to this execution.
+ * @return True to run the [ConditionRef.then] side, false the
+ * [ConditionRef.orElse] side.
+ * @throws Exception on failure; the task instance is marked failed.
+ */
+ @Throws(Exception::class)
+ fun decide(
+ context: Context,
+ client: Client,
+ ): Boolean
+
+ /** Never called: the SDK runs a condition through [decide]. */
+ override fun execute(
+ context: Context,
+ client: Client,
+ ): Unit =
+ throw IllegalStateException(
+ "Condition '${javaClass.name}' runs through decide(), so execute() is
never called",
+ )
+}
+
+/**
+ * A condition registered with a Dag: name the task each outcome runs.
+ *
+ * ```java
+ * dag.If(HasRows.class).then(load).orElse(reportEmpty);
+ * ```
+ *
+ * A named task runs after the condition, so naming it records that edge, as
+ * [Deps.Flow.before] would. The condition skips only the side not taken; what
+ * runs after both sides needs a trigger rule that tolerates one skipped
+ * upstream, such as `none_failed_min_one_success`.
+ *
+ * @see DagDef.If
+ */
+class ConditionRef private constructor(
+ private val ref: TaskRef<Boolean>,
+ private val sides: ConditionDef,
+) : Deps.Flow {
+ companion object {
+ /**
+ * @suppress
+ *
+ * Marks a registered task as a condition. Public so a generated wiring
+ * view can call it; user code reaches a condition through [DagDef.If].
+ */
+ @JvmStatic
+ fun of(ref: TaskRef<Boolean>): ConditionRef {
+ val sides = ConditionDef()
+ require(ref.def.decider == null) {
+ "Task '${ref.def.id}' already decides what to skip; declare it once"
+ }
+ ref.def.decider = sides
+ return ConditionRef(ref, sides)
+ }
+ }
+
+ /** Airflow task ID of the deciding task. */
+ val id: String get() = ref.def.id
+
+ /**
+ * Names the task that runs when the condition holds.
+ *
+ * @param task Task of the same Dag, which the condition skips when it does
+ * not hold.
+ * @return This condition, so [orElse] can follow.
+ * @throws IllegalArgumentException if the task belongs to another Dag, or
+ * this side is already named.
+ */
+ fun then(task: TaskRef<*>): ConditionRef = name("then", task) {
sides.whenTrue = it }
+
+ /**
+ * Names the task that runs when the condition does not hold. A condition
+ * needs no `orElse`; without one, nothing is skipped when it holds.
+ *
+ * @param task Task of the same Dag, which the condition skips when it holds.
+ * @return This condition, for chaining.
+ * @throws IllegalArgumentException if the task belongs to another Dag, is
+ * the `then` task, or this side is already named.
+ */
+ fun orElse(task: TaskRef<*>): ConditionRef = name("orElse", task) {
sides.whenFalse = it }
Review Comment:
Done in 5b57b301175: the sides are Then and Else now, matching If and the Go
SDK.
--
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]
