When I was reading the proposal in the devlist I thought  I understood it but re-reading the AIP I have some questions again. Might be I did not understand the use case fully or the description is drifting into a solution space too early where I am lost.

(1) From the problem description I understand you have a large XCom and want to drive thousands of mapped tasks. You see a large overhead in the DB by mapping/creating and the during processing updating them. Correct?

(2) If I understand (1) correct then if you make a node local mapping/execution then you still would like to track all mapped tasks from the UI respectively, correct? Then I would not understand how the DB load would be reduced except that you save round-trips of scheduler→worker. Still all results (and in a crash a failover to another worker) should be recorded on each mapped task?

(3) If I was wrong in (1) and you want a task local mapping w/o tracking all mappings in the DB then you "just" want to run some mapped tracking within one large task. I think you are free to implement "whatsoever" iterator operator - but if it fails in between then only one atomic state is recorded in the DB and can be seen in the UI and increments of how far the iteration went are not known for the scheduler. So I assume you still want to have the mapping in the UI?

(4) Consequence of (2) would be that a kind of "mini scheduler" would be needed on the worker to track the progress of individual mapped tasks (which would prevent to round-trip to the central scheduler). And of course this would need to have DB access to track/add all tasks... and this is actually one thing we want/need to get rid of in Airflow 3

(5) Another consequence of (2) would be that you have challenges in case of failover (the worker just dies - who is taking over? What about the progress?) as well as you task mapping is mostly limited to the capacity of one node. How would another scheduler take over and know that on a partial local expanded result where to take over?

Can you maybe add a diagram (UML communication flow or so?) to have a better understanding of the current flow and the future/target communication flow that you want to achieve?

Also the problem description introduces with UI problems - how do you think the change will resolve these?

You describe a lot about async, is the desired feature mainly an async polling for thousands of mapped async external tasks or is it "real work" that is computed e.g. on a celery worker with concurrency (=sync) that is to be tracked. Or an universal feature (sync+async)? (I am asking because long term there was also a demand to track results from external sub-workflows in airflow like Cosmos with dbt such that the external orchestrator runs and just reports to Airflow further details but Airflow is delegating the details so something more native - I just believe we have no AIP for this?)

Hello Jens, thank you for your elaborate response.

The proposal I've put down now is indeed already oriented more into the solution I came up with yet it doesn't mean it has to be that way, I just used it to describe what the was the problem we wanted to solve and how we solved it with was already available in Airflow 2.

(1) Yes you are correct.

(2) We personnally don't need the tracking of the individual tasks within the UI, but I think Jarek and Ash would have like to see a more generic solution in which we have both, so local execution of all mapped tasks but also being able to track the progress/state visually within the UI, so that you don't loose that aspect in Airflow when using this alternative approach, which is understandable in a product standpoint.  I don't know how this could be accomplished efficiently without overloading the database, unless you have a distributed postgres behind the scenes with partitioned tables, but maybe there is also another possible solution to solve this.
    
(3) Yes, in our case we don't have the need to individually track the progress of all mapped tasks, so it's an all or nothing, comparable with a for loop within a PythonOperator.  An improvement I haven't implemented yet in case of failure, is to track the already executed mapped tasks through XCom, so that when we would re-run that same iterated operator task again, it would only re-run the failed ones, instead of just re-running them all.  I think this feature is technically feasible.
    
(4) That would be the ideal solution and that's also I think what Jarek was after, but that would mean we would have 2 types of schedulers, the global one and a specialized micro/local scheduler per worker, which by default would always delegate to the global one unless you want the alternative local execution.  In the solution I came up with the local scheduling is done within the IterableOperator, which of course means code duplication for the scheduling aspect, but that problem will always exist even if we would introduce the notion of a local scheduler.  The later would be a better approach, as you could introduced mutiple local schedulers with different implementations.
    
(5) That I described in point 3, I think with the current solution we could keep track of it through a list of mapped indices stored in an XCom, as an XCom is linked to a DagRun, we could retrieve it when re-running the task.  Of course XCom's also need DB access, unless we will have alternative XCom backend providers in the future, we for example use a Kubernetes PVC XCom backend, so XCom's aren't a real problem in our case.

I mention the async aspect a lot, because we use a lot the MSGraphAsyncOperator, which is a deferrable operator, as the msgraph core library from Microsoft is implemented as async only, hence why it's implemented that way but then also gives you the opportunity to apply the producer/consumer pattern and avoids blocking workers while waiting for responses, which in case of paged results can be very handy (I also started a PR to do the same with the GenericTransfer operator), as you can distribute the load between the workers and triggerers.  The solution I've implemented and proposed here is universal, as we use the IterableOperator as well for async operations (IO related) as well as for real multithreaded computing/processing tasks written in pure Python using the PythonOperator.  The current implementation doesn't support @task decorated methods yet nor task groups, the later would be more challenging to implement with the current solution I've done, so that's why a micro/local scheduler would maybe be a better approach.

If you don't do mapped task tracking, you don't have the database/scheduling problem, the UI performance problem is a side effect of the former one.

I'll try to draw a diagram to visually represent the alternative flow.

Just to clarify. I do not want to have "mini" or "remote" scheduler at all. I think that with Airflow 3 architecture, a LOT (if not all) concerns that lead to the proposal will be alleviated, and with certain optimisations, when you mark all mapped  task instances of the same tasks (and in more general case, for example, all tasks belonging to a task group) to be executed on the same machine, you can achieve pretty much the same optimisations you are after - while not introducing a new "scheduling construct" - i.e. kind of sub tasks within a single airflow task. I think this is what the proposal boils down to, and I still think it can be done better from the product perspective.

I think a lot of overhead that is experienced now, will be much, much less in the new Architecture. That is one of the main goals of Airlfow 3 - to get rid of the DB connections overload, unnecessary excessive communication and allow for certain optimisation on the DB side of the things by "concentrating" the DB access in a few central components. Simply speaking the performance gains you see with Airflow 2 will be largely gone in Airlfow 3, because of the architecture. And I would not even consider elaborating too much about this proposal before we see Airflow 3 in action and see how much improvement you can get there, and most importantly, whether similar improvements cannot be achieved by optimising some of the existing flows. For example I can imagine - when we use some kind of "task affinity" - that the communication Task API → Workers might be heavily optimized - for example batched eventually and achieve pretty much comparable performance as when iterating over mapped operators in the same task. If we know that mapped tasks are going to all be executed in a single machine, this is completely no problem (just a bit of engineering effort) to optimize things away:

• send all xcom values as single message
• heartbeat only the "worker instance" not individual tasks within the worker instance
• batch sending status updates back to Task SDK
• and so on

All that could be done at the level of Task SDK API and Airflow code, without impacting the fact that "individual task instances" keep their individual task instance status, can be monitored and individually cleaned /interacted with from the UI etc. 

For me this proposal is ONLY about optimising performance, and we are **just** doing it for Airflow 3 - and IMHO we should build on top of those optimizations rather than looking at the current Airflow 2 performance issues. This is already past. Let's look into the future.

Hello Jarek, then I think it's best to wait until those optimisations in Airflow 3 are done before continuing on this AIP imho?  Are there already branches/PR where we can follow the work that's being done there, because that's really interesting me.

Hi David,

I've also considered that a large volume of XComs could become a performance bottleneck. One idea I had was to implement a custom local XCom backend as a simple and efficient solution. For example, using something like RocksDB as a low-level embedded key-value store for the XCom backend could help eliminate roundtrips to the metadata database and reduce the load caused by storing or setting large amounts of XCom data.

Hope this idea helps!

Hi David Blain after re-reading from a "longer time ago" and also having some use cases in mind I am now 50% with you that a "Lazy Mapping" or "Deferred Operator Mapping" might be useful. I am not sure if compatible but I see the following use cases:

• We had  a medium sized Mapped Task Group with ~6 tasks where Dag Authors attempted to map via a list that had 2000 elements. That makes 12000 tasks in a DAG run and because the list was passed in as a DagRun.conf the scheduler died in OOM becasue it attempted to load 12k tasks with a cross product of the DagRun.conf into memory. Reducing the list to a couple of hundred reduced the problem footprint but still scheduling a very large mapped lis takes excessive time in Scheduler. (Especially in Mapped Task groups as it needs to check many times in the loop next elegible tasks to schedule+queue and looping over the list many many times.
  → So in this case I assume an incremental mapping (e.g. Map always 50 instances, once they are done map the next 50) would be helpful to reduce latency of first tasks landing as well as reduced scheduler overhead in looping over a big list of open tasks. This would solve our current problem that in Mapped Task Groups we are not sure if we can get very much over the default limit of 1024. But we have demand to make it rather 10k or 100k tasks in a very long and big flow.
• One other Use Case we have seen is also a rather "Non Dag" case where actually we are lagging an option to implement a Loop in a Dag. Assume you make a model training and based on the results you need to decide if you run another training epoch. Basically you want to loop the ML pipeline until either you have good results or you give-up the experiment. Today you would need a driver Dag that controls a training Dag that calls in a loop.
  → How about if such mapping is "mis-used" ==leveraged to define the Mapped Task Group as the Loop base. You run one cycle and at the end of the group you check for results. If not good enough you add one element to the mapping such that another set is scheduled. At the end after N loops you still have all tasks with logs and XCom and you can still via the UI inspect the overall duration and results. The only "problem" is that the "Map Index" is mis-used to be a kind of "Loop Iteration Counter".
  Does also not mean that the Mapping at the end needs to be made via a distinct XCom but assuming there might be a generated task in the Mapped Group which implements an interface to check and potential add elements to mapping as increment

Jarek Potiuk in Devlist you also had a "loud thought" about support for non Dag type workflows for have loops. Would the second point be in a direction you thought about?

So TLDR, do we maybe need to re-phrase the AIP for "Late Mapping" which can b used for "Lazy Mapping", "Incremental Batch Expansion" and as well as "Mapping for Workflow Loops"?