Introduce ExecutionMode.WATCHER to reduce DAG run time by 1/5

[tatiana]

Introduce a new high-performance execution mode, named ExecutionMode.WATCHER, following the implementation of the producer and consumer operators in PRs #1982, #1993 and #1998.

Initial performance analysis indicates that this mode will reduce the total DAG Run time to execute some dbt pipelines in Airflow to 1/5 of the original time. For example, if a Cosmos DbtDag takes 5 minutes to run with the default ExecutionMode.LOCAL, it will now run in 1 minute with the new ExecutionMode.WATCHER. The performance gain will be comparable to running the dbt CLI locally.

In the near future, there will also be benefits related to CPU and memory utilisation, as users will be able to run the producer task on a more powerful node with increased CPU and memory resources. In comparison, the consumer nodes can have less CPU and memory. Further development (#1972 and #1973), testing, and analysis are needed to evaluate this.

Context

As of Cosmos 1.10, when users leverage the default ExecutionMode.LOCAL, each dbt model becomes an Airflow run task, and dbt is invoked in each of those tasks.

We noticed that the cost to run the same pipeline with plain dbt core varies significantly by running:

• the whole dbt command using a single command
• running one dbt command per model

For example, for the https://github.com/google/fhir-dbt-analytics project, these numbers were, on average, 5 minutes and 30 seconds (by running a single dbt run for the whole pipeline) versus 32 minutes (when using 184 dbt run commands as illustrated in https://gist.github.com/tatiana/c7831173ab09bf05d88839fb0b557920).

Similar to the ExecutionMode.AIRFLOW_ASYNC, this mode aims to reduce the number of times the dbt command is invoked, while still allowing users to have observability of the dbt workflow via the Airflow UI and being able to retry individual tasks.

Overall solution

• Use existing Cosmos DAG rendering techniques - implemented in this PR
• Have a single Airlfow task to run "all the pipeline" (selected by the user) - implemented in Add DbtProducerWatcherOperator for the proposed ExecutionMode.WATCHER #1982
• Use dbt Core callbacks https://docs.getdbt.com/reference/programmatic-invocations#registering-callbacks to track how the model's execution is progressing and update different Xcoms (one Xcom per model) - implemented in Add DbtProducerWatcherOperator for the proposed ExecutionMode.WATCHER #1982
• All the other tasks, by default, should watch their designated Xcom - implemented in Add DbtConsumerWatcherSensor for the proposed ExecutionMode.WATCHER #1998 and used in this PR

This proposal follows up on a successful internal PoC (https://github.com/astronomer/oss-integrations-private/issues/185), available in the branch https://github.com/astronomer/astronomer-cosmos/tree/single-run-execution-mode.

Benefits

An initial performance analysis by @pankajkoti showed promising results:

Experiment	Number of threads	Execution time (s)
dbt build	4	6 - 7
dbt run for each of model locally		30
Cosmos default ExecutionMode.LOCAL in Astro CLI locally		10 - 15
Cosmos proposed ExecutionMode.WATCHER in Astro CLI locally	1	26
	2	14
	4	7
	8	4
	16	2
The ExecutionMode.WATCHER in Airflow with an Astro deployment	8	5

Example of usage

Example of DAG topology, with the producer task preceding the others.

The dbt root nodes are set with trigger_rule always, so they start sensing once the producer begins.

Producer task runs dbt Core, as shown on the logs:

Consumer task senses XCom, waiting for producer to finish running dbt:

Evidence that producer is running concurrently to the dbt root nodes sensing:

Related tickets

Closes #1964
Closes #1959 (*)

(*) I ended up implementing this while trying to enforce the producer task to run before the consumer tasks when running airflow dags test.

Outside of the scope of this PR:

• Documentation (this will be added as part of Build base images for running dbt in Docker #245)
• We are not implementing support for the following operators in the ExecutionMode.WATCHER mode:
  • LS
  • Run operation
  • Docs
  • Clone
    Since it does not make sense to have them, we can review them later.

There are many other tasks related to this execution mode that can be tracked by searching issues using label:execution:watcher:
https://github.com/astronomer/astronomer-cosmos/issues?q=is%3Aissue%20state%3Aopen%20label%3Aexecution%3Awatcher

Credits

The idea for this approach appeared in a discussion with @ashb.

The implementation of this feature is the result of teamwork with @pankajastro and @pankajkoti, both directly and indirectly involvement via PoC and previous PRs:

• Co-authored-by: Pankaj Koti pankaj.koti@astronomer.io
• Co-authored-by: Pankaj Singh pankaj.singh@astronomer.io

[netlify]

✅ Deploy Preview for sunny-pastelito-5ecb04 canceled.

Name	Link
🔨 Latest commit	3d1f5b0
🔍 Latest deploy log	https://app.netlify.com/projects/sunny-pastelito-5ecb04/deploys/68e3ff359dbbb80008a9e003

[pankajkoti]

It's looking great to me. I don't have any additional comments at this point, apart from the render_config question for the producer task. Hope we can merge this soon! 🤞🏽

[pankajastro]

Looks great, @tatiana!!

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.83%. Comparing base (5f6e710) to head (3d1f5b0).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1999      +/-   ##
==========================================
+ Coverage   97.81%   97.83%   +0.01%     
==========================================
  Files          87       87              
  Lines        5542     5591      +49     
==========================================
+ Hits         5421     5470      +49     
  Misses        121      121              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull Request Overview

This PR introduces ExecutionMode.WATCHER, a new high-performance execution mode for Cosmos that aims to reduce DAG run time by 80% (from 5 minutes to 1 minute) by using a producer-consumer pattern instead of running separate dbt commands for each model.

Key changes:

• Adds ExecutionMode.WATCHER enum value and PRODUCER_WATCHER_TASK_ID constant
• Implements watcher operators (DbtProducerWatcherOperator, DbtConsumerWatcherSensor, and specialized operators for different dbt commands)
• Integrates watcher mode into the Airflow graph building process with producer task creation and dependency management

Reviewed Changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
cosmos/constants.py	Adds WATCHER execution mode enum and producer task ID constant
cosmos/operators/watcher.py	Implements all watcher operators including producer, consumer sensor, and command-specific operators
cosmos/airflow/graph.py	Integrates watcher mode into graph building with producer task creation and dependency setup
dev/dags/example_watcher.py	Provides example DAG demonstrating watcher execution mode usage
tests/operators/test_watcher.py	Adds comprehensive integration test for watcher DAG functionality
tests/dbt/test_graph.py	Updates hash value for macOS compatibility in dbt cache testing

Comments suppressed due to low confidence (1)

cosmos/operators/watcher.py:1

• Using raw string literal r\"source\" is unnecessary here since there are no escape sequences. Should be just \"source\"."

from __future__ import annotations

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[yetudada]

I'm so excited to see this PR! Well done @tatiana @pankajkoti @pankajastro!