Hi all,
I'd like to propose a way for a task's trigger condition to refer to the
specific state of an upstream task, rather than counting states across all
upstream tasks.
Business use case
We run a multi-step batch pipeline with the following task graph:
task_a >> task_b >> task_c
task_a >> task_d
task_b >> task_d
task_c >> task_e
task_d >> task_e
task_d >> task_f
task_e >> task_f
The scheduling conditions required are:
Task
Condition
task_d
task_a = success AND task_b = failed
task_e
task_c = success OR task_d = success
task_f
task_d = failed AND task_e = success
These conditions describe when each task is eligible to be scheduled, not what
happens inside the task:
* task_d has two upstream tasks. It must run only when one specific
upstream succeeded and the other specific upstream failed — not when any
upstream failed, and not when both failed.
* task_e has two upstream tasks. It must run when either one of them
specifically succeeded — not when any upstream in the whole graph succeeded.
* task_f has two upstream tasks. It must run when one specific upstream
failed and the other specific upstream succeeded — opposite state requirements
on each named upstream simultaneously.
In all three cases the condition is a precondition on scheduling, not on
execution. The scheduler already evaluates preconditions before dispatching a
task — trigger rules are exactly that mechanism. The missing capability is that
trigger rules count states across the whole upstream pool anonymously; they
cannot test the state of a specific named upstream task.
A BranchPythonOperator or inline Python check cannot substitute here — it runs
only after the task is already scheduled and started, and it does not affect
how the scheduler treats downstream tasks that depend on the outcome.
Why existing trigger rules cannot express these
All 13 trigger rules treat upstream tasks as an anonymous pool — they count or
ratio states across the whole group. They have no concept of which task is in
which state. The moment a condition requires naming a specific upstream task,
every existing rule breaks down.
Attempting task_d (task_a = success AND task_b = failed):
* one_failed — catches task_b = failed ✓, but also fires when task_a itself
failed ✗.
* all_done — fires for every combination of outcomes, including both
succeeded or both failed ✗.
* none_failed — fires only when nothing failed, the opposite of what we
need ✗.
No existing rule can say "this specific upstream must succeed AND that specific
upstream must fail."
Attempting task_e (task_c = success OR task_d = success):
* one_success — fires when at least one upstream succeeded, but task_e has
upstream tasks beyond task_c and task_d in the graph. one_success fires as long
as any one of them succeeded, so it fires even when neither task_c nor task_d
specifically succeeded ✗.
* all_success — requires every upstream to succeed, far too strict ✗.
* all_done — fires for every outcome combination ✗.
No existing rule can express OR across two specific named tasks while ignoring
the states of all other upstreams.
Attempting task_f (task_d = failed AND task_e = success):
* one_failed — fires when any upstream failed, not specifically task_d ✗.
* none_failed — blocks entirely when task_d failed, which is exactly the
case we want to allow ✗.
* all_done — fires regardless of which tasks succeeded or failed ✗.
No existing rule can combine a required failure on one named task with a
required success on another.
The fundamental problem: trigger rules count states across the whole upstream
pool. As soon as the condition requires knowing which task is in which state,
every rule either over-fires or under-fires.
Current workaround
Insert a gate task with trigger_rule="all_done" that reads upstream states at
runtime (via RuntimeTaskInstance.get_task_states, which is not documented as
public API), evaluates the expression, and raises AirflowSkipException when it
is false. It works, but:
* every condition adds an extra task to the graph and to scheduling
* the real dependency logic is hidden in Python instead of shown in the DAG
* it relies on internal APIs that may change
* an OR condition can't fire early, because the gate always waits for every
upstream task to finish
In the pipeline above, three gate tasks would be needed — one per conditional
task — tripling the graph complexity for logic that should be expressible in
the DAG definition itself.
Relationship to AIP-106
AIP-106 (Composable Trigger Rules) adds count and ratio expressions over
upstream states, which is a different problem. It explicitly lists per-upstream
named conditions as out of scope and a possible separate AIP. This proposal is
meant as that follow-up and could reuse the same TR namespace.
Possible API-Draft
from airflow.sdk import TR, success, failed
task_d = MyOperator(
task_id="task_d",
trigger_rule=TR.when(success(task_a) & failed(task_b)),
)
task_e = MyOperator(
task_id="task_e",
trigger_rule=TR.when(success(task_c) | success(task_d)),
)
task_f = MyOperator(
task_id="task_f",
trigger_rule=TR.when(failed(task_d) & success(task_e)),
)
The scheduler would evaluate this in the trigger-rule dependency check, using
states it already has. Named tasks would automatically become upstream tasks.
I am preparing a full AIP draft and will share the Confluence link once my wiki
account is approved.
In the meantime, I am happy to share the draft document directly if that helps
the discussion.
Thanks,
Gowthami
