Status

Short summary

Allow Triggerers to yield multiple events in a single trigger run, enabling true streaming-style workflows for async operators

This AIP is a proposal to add some kind of streaming support in Airflow through triggers, that way if an operator can be run in an async way which yields (a lot) of results (e.g. TriggerEvent's), we can simulate the effect of streaming in Airflow and thus also allow lazy task expansion as tasks will get expanded as new events are being yielded.  The current proposition differs completely from the initial one, and also doesn't cover all aspects like for example an XCom being iterable which is not supported at the moment, for that to work we would need another mechanism not dependant on the triggers.

This AIP is a proposition to support an alternative way of expanding multiple XCom’s on operators without the need to know how may tasks will be expanded in advance, which is the case with streams or iterables.

That’s why I would also like to introduce then notion of streamable XCom's.  In the current Airflow implementation, only list and dicts are supported as XCom collection types on which can be expanded on (e.g. SchedulerDictOfListsExpandInput and SchedulerListOfDictsExpandInput).

I would like to add support for iterable's (e.g. lazy evaluation collection).  The reason why I would like to introduce the support of iterables is to allow the implementation of streamable XCom's.  In case of the HttpOperator or the MSGraphAsyncOperator for example, when those return pages results, the operator depending on the results of that operator needs to wait until all pages have been loaded before being able to process them.

This has 4 disadvantages:

• performance, as depending operators wait until all pages have been loaded before being able to start
• this also means the iteration is done twice as opposed to a streaming solution as there both steps (e.g. tasks) could be done within the same iteration
• memory usage, as all pages have to be loaded into memory as a materialized XCom before the next operator can consume it
• as XCom's have to be able to determine their length in advance, it's impossible to implement the filter functionality (like map or zip or concat) in the current implementation, as filter will dynamically affect the length of the returned XCom length.

, HttpOperator/MSGraphAsyncOperator). This reduces repeated context switching and "ping-pong" between scheduler → worker → triggerer, enabling efficient in-trigger pagination and lazy task expansion.

Image AddedImage AddedImage AddedImage AddedImage AddedImage Added

Motivation (why)

• Many async operators implement pagination by deferring to triggers and then reentering the operator repeatedly. Even with start_from_trigger, the current system processes only the first TriggerEvent yielded; additional pages require repeated defer/resume cycles.

• Repeated context switches create scheduler/worker/triggerer overhead and slow throughput for high-volume paginated APIs.

• A streaming-capable triggerer can yield many events in one run; letting Airflow process those events (without full defer/resume for each page) produces substantial performance gains and allows lazy expansion of downstream tasks as pages arrive.

Goals

1. Allow triggers to yield many events during a single run and have those events processed incrementally by the operator runtime/scheduler without repeated operator deferral cycles.

2. Allow operators that know how to consume events to expand mapped tasks or XCom-driven iterables lazily as events arrive.

3. Preserve serializability guarantees (no unserializable callables in trigger args).

4. Be backward-compatible: existing triggers/operators continue to work.

Non-goals

• Making arbitrary XComs iterable across the cluster. (That requires additional XCom semantics; out of scope.)

• Allowing triggers to carry non-serializable callables/closures. Triggers must remain serializable.

Demo

High-level design

Key idea

Extend the triggerer-to-scheduler event path so that a single trigger run may produce a stream of events which the scheduler can incrementally deliver to the operator context (worker/scheduler) and allow the operator to handle each event without a full defer/resume roundtrip per page.

Components & responsibilities

• Triggerer: can yield multiple TriggerEvents during run(). It still must be serializable. If the trigger performs pagination, it should do that inside its run() loop and yield each page as an event.

• Scheduler: must accept and process multiple events from the same trigger-run. Instead of ignoring all but the first yielded event, scheduler accepts the events and binds them to the operator’s next_method handler incrementally.

• Operator / Worker:

  • Operators supporting streaming implement a handle_trigger_events(events: Sequence[TriggerEvent], context) callback (or extend next_method to accept a batch/stream), which may:

    • process events and optionally expand mapped tasks or create XComs,

    • decide whether trigger should continue producing (e.g., requests more pages) — but triggers are the ones that control pagination inside run().

  • Operators still raise TaskDeferred to start the trigger if they cannot proceed synchronously.

• Start-from-trigger: still used — when present we skip the initial worker-run step. The AIP assumes start_from_trigger exists and is used (Airflow 3.0+).

API changes / additions

1. Triggerer run behavior: no signature change, but scheduler processing semantics change: treat yield as a stream of events, not just the first.

2. Scheduler/Triggerer protocol:

   • Add an event envelope with fields {trigger_run_id, sequence_index, payload, is_last} for each yielded event. This helps ordering and safe incremental processing.

3. Operator callback:

   • Add next_method(self, event: TriggerEvent | List[TriggerEvent], context) accept either single or batched events, or an optional handle_trigger_events(self, events, context) method.

   • For backward compatibility, if operator implements the old next_method(event, context), the scheduler will call it for each event individually.

4. DAG parsing checks: if start_from_trigger is enabled and start_trigger_args contains a non-serializable callable, raise at parse time (existing check you already mentioned).

Ordering & consistency concerns

• Use sequence_index to process events in order from the same trigger run.

• If operator processing of an event fails, the scheduler should:

  • surface the failure (normal task retry semantics) and

  • stop accepting further events for that run until retry/resume behavior is resolved.

• If the trigger signals is_last, scheduler can mark the trigger-run completed.

Security / serializability

• Triggers and all trigger args must remain serializable (no lambdas/closures).

• DAG parsing validates start_trigger_args types and raises early if a non-serializable callable is present (your existing behavior).

Backward compatibility / migration

• Existing triggers/operators continue to work: scheduler defaults to single-event processing if operator doesn’t opt-in.

• Operators that want streaming implement the new batched handle_trigger_events or accept repeated next_method calls. Provide an adapter shim that calls an operator’s single-event next_method repeatedly if the operator hasn’t implemented streaming (so no breakage).

• Update common async operators (MSGraphAsyncOperator, HttpAsyncOperator) to implement in-trigger pagination and yield many TriggerEvents.

Tests & acceptance criteria

• Unit tests for:

  • multiple-event processing ordering,

  • failure during mid-stream event handling,

  • serializability check during DAG parse.

• Integration tests:

  • MSGraphAsyncOperator with 10+ pages yields vs old approach measured in scheduler/worker/triggerer traces (assert fewer context-switch cycles).

• Performance benchmarks demonstrating reduced scheduler <-> worker <-> triggerer hops.

Example operator migration notes

• Refactor operator to:

  • Move pagination into the trigger run() loop,

  • Yield each page as a TriggerEvent envelope,

  • Ensure start_trigger_args are serializable,

  • Implement handle_trigger_events to consume events, create XComs, and expand tasks lazily.

Open questions / future work

• Native iterable XComs (separate AIP).

• Fine-grained backpressure between triggerer and scheduler (if extremely high event rates).

Event batching strategies (time / size) to trade off latency vs. scheduler invocation overhead.

Sequence diagrams

Below are two diagrams:

• Current/present behavior (with start_from_trigger but only first yielded event processed leading to repeated defer/resume cycles).
• Proposed behavior (trigger yields many events in one run; scheduler forwards them incrementally to the operator; fewer context switches).

Image AddedvsImage Added

Proposal

Possible workaround?

You could argue that you could write a PythonOperator or a task decorated method in which you loop over the multiple inputs from the XCom to pass as an argument to the operator or even a hook.  While the later would be a valuable solution, the first one wouldn’t as it’s a bad practise to execute an operator from within a PythonOperator, see the discussion about this topic on the devlist.

There is already an safeguard implemented for this which checks if an operator is executed from a PythonOperator, and if so, logs a warning stating an operator cannot be called outside of a TaskInstance.  In the future this will probably become prohibited and will raise an AirflowException in that case.

But let’s hypothetical assume this would still be allowed, how would you loop the inputs from an XCom to an operator if that operator is deferrable?  You would need to take multiple aspects into account.

First of all, you would need to catch the raised TaskDeferred exception, as this is how a deferrable operator works.  The TaskDeferred exception contains the triggerer associated to the deferred operator to be executed, which behinds the scenes returns an async generator.  This means that you will have to cope with the event loop of asyncio to be able to run the async triggerer from your PythonOperator, as the later one isn’t executed in an async way.  Maybe this is also a good moment to start thinking of natively supporting async method’s in the PythonOperator without worrying about coping with the event loop (e.g. PythonTriggerer)?

Next to that, once you achieved to execute the triggerer, you will also have to check if a next_method was specified, which has to be executed on the deferred operator once the trigger has completed.

And last but not least, the execution of the next_method could also re-raise a TaskDeferred exception if the deferred operator implements the producer/consumer pattern, which means you’ll have to take into account recursion.  For example the MSGraphAsyncOperator implements the producer/consumer pattern in such a way that the worker triggers the request to the MS Graph API, but instead of blocking the worker waiting for the response to arrive, releases it and delegates it to the triggerer, avoiding blocking workers unnecessarily.  That way when the triggerer receives the response, it gives the received response back to the operator (e.g. worker) without blocking the worker while awaiting for the response.

That’s already a lot of technical challenges you have to solve if you want to execute a (deferrable) operator from within a loop in a PythonOperator.

Beside that, you maybe would also like to introduce some multithreading to speed up the processing instead of just looping in a sequential manner, unless you want sequential execution in the given input order, which is also a issue raised in this discussion.  But there you will also have to be careful, because you just can’t execute a deferable operator having async code through a ThreadPoolExecutor.

Image Removed

What problem does it solve?

Faster execution of downstream operators if the initial operator is returning a paged XCom result.

Image Removed

In the screenshot above, you can see the new lazy expandable task mapping implementation in Airflow 3. It demonstrates how the MSGraphAsyncOperator returns a deferred iterable (as an XCom), allowing downstream tasks like SQLInsertRowsOperator to expand lazily as the scheduler (e.g. TaskMap) iterates over the paged results.

This means the SQLInsertRowsOperator does not have to wait for all pages to be fetched before expanding into mapped tasks. Instead, the MSGraphAsyncOperator initially returns only the first page, and as the scheduler expands the mapped tasks, it fetches additional pages on demand.

This approach offers significant improvements over the traditional model, where the MSGraphAsyncOperator would have to fetch all pages upfront before expansion could begin—resulting in slower performance and higher memory usage.

Why is it needed?

Improves performance and significantly reduces waiting times.

Are there any downsides to this change?

With the current proposal, task expansion is fully managed by the scheduler using the TaskMap. Traditionally, this requires knowing the length of the XCom in advance in order to expand the mapped task. To support lazy XCom evaluation, however, the XCom must now be resolved by the scheduler rather than the operator. The key advantage of this change is that the scheduler no longer needs to know the total length of the XCom upfront, enabling partial or streaming-style task expansion.  This means the scheduler won't create mapped tasks instances anymore, but fully unmapped task instances, hence why it has to be resolved with the scheduler as opposed to the original implementation.  A downside in the current POC is that the evaluation of the deferred itrable XCom is being done within the scheduler instead of the worker/triggerer, but maybe there we can check with Unknown User (ash) how this could be improved. 

In this model, the scheduler initially expands only the first n task instances from the MappedOperator. The remaining task instances are expanded asynchronously by the TaskExpansionJobRunner, which allows new tasks to be created incrementally as more data becomes available. Meanwhile, the executor can begin running the already-expanded tasks immediately, enabling a producer-consumer style execution and improving overall throughput.

This leads to a more dynamic and efficient execution model, especially useful for deferred or paginated iterables, where loading all data upfront could be slow, memory-intensive, or infeasible.

Which users are affected by the change?

None

How are users affected by the change? (e.g. DB upgrade required?)

None

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

None, as this is just an alternative way of expanding XCom's on a operator, instead of calling the existing partial method of the MappedOperator, you can now call the iterate method.