Asset-triggered Dag runs cause scheduler memory spikes
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Under which category would you file this issue?
Ariflow Core
### Apache Airflow version
`main`
### What happened and how to reproduce it?
When the scheduler creates an asset-triggered Dag run, it selects every `AssetEvent` in the event window as a full ORM object. This materializes each event's JSON metadata and relationship state in the scheduler process before writing rows to `dagrun_asset_event`.
The scheduler's peak memory therefore grows linearly with the number of events. A production-path benchmark using a 512-byte JSON payload per event produced the following baseline results:
| Database | Events | Duration | Python peak | RSS increase |
|---|---:|---:|---:|---:|
| SQLite | 1,000 | 0.537 s | 6.23 MB | 0.17 MB |
| SQLite | 10,000 | 1.785 s | 54.24 MB | 79.98 MB |
| SQLite | 50,000 | 12.545 s | 274.21 MB | 465.98 MB |
| PostgreSQL | 1,000 | 0.339 s | 7.39 MB | 1.88 MB |
| PostgreSQL | 10,000 | 2.510 s | 62.34 MB | 108.78 MB |
| PostgreSQL | 50,000 | 10.046 s | 289.72 MB | 483.77 MB |
| MySQL | 1,000 | 2.622 s | 6.76 MB | 0.75 MB |
| MySQL | 10,000 | 4.478 s | 54.54 MB | 98.21 MB |
| MySQL | 50,000 | 21.502 s | 272.63 MB | 475.28 MB |
Reproduction outline:
1. Create an asset-scheduled Dag with `catchup=False`.
2. Insert 1,000, 10,000, or 50,000 matching `AssetEvent` rows containing a 512-byte JSON value.
3. Add the corresponding `AssetDagRunQueue` row.
4. Invoke `SchedulerJobRunner._create_dag_runs_asset_triggered`.
5. Measure elapsed time, Python allocations, and scheduler RSS during Dag run creation.
Fixture creation is outside the measured interval. Every run creates the expected number of
`dagrun_asset_event` associations.
### What you think should happen instead?
Dag run creation should preserve exact consumed-event membership without materializing complete `AssetEvent` objects in scheduler memory. The association can be written as a set-based `INSERT ... SELECT` in the existing transaction, using the same direct-asset, alias, and event-window predicates.
This keeps Dag run creation, event association, and queue consumption atomic while making scheduler-side Python memory effectively independent of the number and payload size of matching events.
### Operating System
_No response_
### Deployment
None
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
The behavior is reproducible on SQLite, PostgreSQL, and MySQL.
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Locate SchedulerJobRunner._create_dag_runs_asset_triggered and trace the existing asset, alias, event-window, and queue-consumption paths. Reproduce the benchmark across SQLite, PostgreSQL, and MySQL, then verify that exact dagrun_asset_event membership and atomic queue consumption are preserved while scheduler memory no longer grows with event payloads.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- mysql, postgresql, python, sqlite
- Domain
- backend, data-engineering, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100