A couple of early comments. 1) This is the hottest of hot paths of the scheduler, so we need to be very careful of every change we make here. 2) Can you please give _real_ examples of where you want to use this? I’m assuming this is driven by some specific workflow you want to handle, and it would help to talk about real cases, not “task a, b, c, and d”. 3) Much like I said to Shahar’s AIP 106 https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/421956843/AIP-106+Composable+Trigger+Rules?focusedCommentId=451979867#comment-451979867: "Why does that need to be a new trigger rule? Couldn't this be an all_done/etc, and the task itself evaluates the upstream results and fails if not?”
The same comment applies here I feel. (This is not a “no”, this is “this is a much bigger change that it might seem at first, so you have to make a well argued case for it) -ash > On 11 Nov 2026, at 08:48, Chandra Gowthami, C via dev > <[email protected]> wrote: > > 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 > > --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
