AIP-115 On-Demand Task Sections

Status

Current Status

It was discussed in the Airflow Dev call that this AIP is better serviced as a plugin and a custom operator instead. https://github.com/apache/airflow/pull/69148 will now allow task instance based scope allowing for custom implementation of the same.

Summary

This proposal introduces a first-class way to define optional task subsections that are skipped in normal Dag runs but can be run later, on demand, for a specific Dag run via a UI/API action.

The motivating scenario is a large scheduled Dag containing expensive, slow, risky, or rarely needed work, where the main run should finish normally while keeping a discoverable, auditable path to run the optional part later for the exact run that needs it.

The user-facing API is a provider operator, OnDemandSectionOperator, placed at the start of the optional path. During a normal run it succeeds and skips its downstream section using Airflow's existing skip mechanism — no new  scheduler behavior and no new task-instance state. An authorized user can later run that section for a chosen Dag run through a dedicated, preview able task-instance action.

Demo

onDemandDemo.mp4

Motivation

On-demand sections are uniquely useful only when all three of these hold at once:

1. The optional work depends on artifacts/state that this specific run produced (so you can't cleanly move it to a separate Dag without re-deriving or re-plumbing those inputs)
2. The go/no-go decision is human and arrives later — minutes, hours, or days — so you can't block the run waiting (rules out HITL) and can't decide at runtime (rules out branching).
3. The work is expensive, risky, or rarely needed, so running it automatically every time is wrong

Airflow supports adjacent workflows today, but none directly expresses *non-blocking work that stays available on demand*:

• Human-in-the-loop (HITL) tasks deliberately keep a Dag run open while awaiting a response. That is the wrong shape for "optional" work — the run should be free to finish.
• Branching / short-circuiting can omit work based on runtime logic, but do  not communicate that a user may run the omitted section later, and expose no first-class action to do so.
• Manual clearing can rerun part of a Dag, but is too low-level: users must understand the graph shape, select the correct task instances, and set downstream options themselves. It is not discoverable from the authoring API.

Other systems (e.g., GitLab manual jobs) expose this concept directly. Airflow should have an equivalent expressed in Dag terms: an operator that makes the intent explicit in the Dag definition and a scoped action in the UI and API.

Example: Gitlab Pipelines

Real World Examples

•   Promote this build to production (CI/CD gate).
  • Build → test → publish immutable artifact → deploy to staging → verify staging, all automatic. deploy_to_production + smoke tests are the on-demand section. A release manager reviews staging and promotes that exact artifact hours later.
  • Why nothing else fits: A separate "deploy-prod" Dag forces you to plumb the commit SHA / artifact version and risks deploying the wrong build. Rebuilding is non-deterministic and slow. HITL would block the run for the entire review window. This is exactly GitLab's manual-job / manual-gate pattern, which Airflow has no clean equivalent for. 
• Regulated "publish final / lock the books" step.
  • Nightly close produces preliminary numbers automatically. An optional "post adjustments → publish final → lock period" section runs only after accounting signs off, on that specific period's run.
  • Why it fits: High-impact, must be explicit, permissioned, and audited — which maps directly onto the AIP's "mutating action + existing  permission + action logging" model. The final numbers must derive from that period's computed state, so a separate Dag would duplicate the whole close. 
• Productizing a "manual clear" runbook
  • Today, when a nightly job's data-quality check trips, on-call follows tribal knowledge: "find the run, clear these five tasks with downstream selected, but not the join." The on-demand section turns that into an authored, labeled, previewable, permissioned button scoped to the failing run.
  • Why it fits: Clearing technically already works, so this isn't new capability — but the discoverability + preview + permission +  authored-intent is a real operational win.
• Optional expensive reprocessing tied to a partition
  • Daily ingest + validate runs automatically. If an anomaly is flagged, an analyst kicks off a heavy full re-derivation of downstream aggregates for that logical date's partition only, on demand.

Note:  if the optional work is independent of the run's own state, a separate parametrized Dag is the better tool

Goals

• Mark an optional subsection as on-demand-only from the Dag definition.
• Let scheduled, manual, and backfill runs complete without waiting for it.
• Identify on-demand sections distinctly in the UI.
• Provide a clear, previewable Run action scoped to a specific Dag run.
• Preserve normal execution semantics (retries, logs, callbacks, pools, queues, XComs, executor behavior, observability, auditability) once run.
• Keep the first implementation small — reuse existing state and dependency machinery, no scheduler rewrite, no schema migration.

Terminology

• On-demand section marker: the "OnDemandSectionOperator" task that starts a optional subsection.
• On-demand section: the downstream tasks controlled by the marker.
• Default skip: the normal behavior in which the marker succeeds and the section is skipped without blocking the run.
• Running the section: the explicit, run-scoped action that clears the section's task instances and lets them execute through the normal scheduler path.

User-Facing Behavior

OnDemandSectionOperator

Dag authors place an OnDemandSectionOperator at the start of an optional path and may give the section a user-facing label:

from airflow.providers.standard.operators.on_demand import OnDemandSectionOperator

production_release = OnDemandSectionOperator(
    task_id="production_release",
    label="Deploy this release to production",
)

verify_staging >> production_release >> deploy_to_production >> smoke_test

• label identifies the action to users in the UI and API. When omitted, the task display name or `task_id` is used.
• Boundary is controlled by `ignore_downstream_trigger_rules` (default `True`): by default the section controls all downstream descendants of the marker. Set it to `False` to limit the section to direct downstream tasks and let later descendants follow their own trigger rules.

The operator is a thin, identifiable subclass over Airflow's existing skip mechanism (`SkipMixin`), analogous to `ShortCircuitOperator`. It is kept as a dedicated operator — rather than reusing a generic `ShortCircuitOperator` — so the UI and API can reliably identify on-demand sections and compute their boundaries from the operator's identity.

For scheduled, manually triggered, and backfill Dag runs:

• The required path runs normally.
• The on-demand section marker succeeds.
• The tasks controlled by the section are skipped by default
• The Dag run can reach a terminal state without waiting for user input or optional work.

Backfills and manual runs therefore do not unexpectedly execute expensive or high-impact optional sections. The section remains visible in the graph and retains a clear action for the specific Dag run.

Why Not trigger_rule="manual"?

trigger_rule currently describes how upstream task states are evaluated. It answers questions such as "did all upstream tasks succeed?" or "did at least one upstream task fail?"

Manual-only execution is not an upstream-state predicate. It is a scheduling policy: the task should not be automatically scheduled unless a user explicitly arms it for this Dag run.

Using trigger_rule="manual" would overload an existing concept and would likely be confusing around joins, branching, setup/teardown tasks, and future trigger-rule additions. This AIP recommends avoiding trigger_rule="manual" as the initial API.

Why Not OnDemand=True in BaseOperator?

• Moves the work into the scheduler. No task runs to do the skipping, so the scheduler must gate the task itself — new scheduling-policy logic on its hot path (the same reason trigger_rule="manual" was rejected).
• Loses the free, reversible skip. #71050 gets skip-and-unskip for free from SkipMixin + NotPreviouslySkippedDep; a scheduler-made skip has no signal to delete, so you'd have to invent a new dep or state.
• Core contract change, not a provider feature. A BaseOperator kwarg lands in every operator's constructor + serialization, essentially forever.
• Every task type must define "onDemand" semantics — setup/teardown, mapped, sensors, branches, deferrables — vs. one operator defining it once.
• Saves nothing on the hard part: "run later" still requires clear-TIs + re-queue the run regardless.

UI/API Proposal

UI

Graph view and the task-instance details view identify on-demand section markers by operator identity and, for a successful, non-mapped marker, expose a primary "Run on-demand section" action.

Selecting it shows a confirmation dialog that lists the affected task instances (from the dry-run preview), accepts an optional note, and offers the "protect running tasks" option, then refreshes.

Public API

Add a focused endpoint, for example:

  POST /dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/runOnDemandSection

The response should include the affected task instances, similar to clear-task-instance responses.

Errors:

• 404 Dag run, task, or marker task instance not found.
• 400 target task is not an on-demand section marker.
• 409 marker is not yet `success`, or an affected task instance is running and `prevent_running_task` is set.
• 403 caller lacks the required permission.

Alternatives Considered

• Dag-run-scoped state store + ShortCircuitOperator: A proposal to record a  per-run key (dag_id, run_id, key) that a ShortCircuitOperator reads(absent/False → skip, True → run), avoiding a new operator. This does not  stand on its own: the gate executes early, so by the time a user clicks "run,"  the gate has already finished and the run has often completed — the section  still has to be cleared and re-run. It also gives the UI/API no reliable way to identify gates or compute section boundaries. This proposal keeps a dedicated,  identifiable operator and performs the clear-and-rerun explicitly; a persistent  store for audit metadata remains a possible future addition.
• ApprovalOperator/ HITL. Parks the run in an awaiting-input state and keeps it unfinished — wrong for optional, non-blocking work.
• BranchOperator/ShortCircuitOperator directly Can omit work but offer no first-class, discoverable action to run it later, and are not identifiable as on-demand sections.
• Manual clearing: Too implicit and not discoverable from the authoring API.
• trigger_rule="onDemand": Rejected — on-demand execution is a "scheduling policy", not an upstream-state predicate. Overloading trigger rules would be  confusing around joins, branching, and setup/teardown.