data-aware scheduling makes wrong dataset updates and downstream dag runs mapping when there are multiple updates during the execution of downstream dag
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
2.10.5
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
I created a demo procedure where the upstream dag updates the dataset continuously, while the downstream dag just sleep 30s to simulate some time-consume tasks.
however, the dataset update and downstream dag run mappings are wired. it looks like below and there would be dangling dataset updates if I stop the upstream dag.
the dangling dataset updates are actually processed by downstream dag but it showed in ui that it would never be processed.

### What you think should happen instead?
### How to reproduce
```
from datetime import datetime as datetime
import airflow
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.operators.python import PythonOperator
from airflow.operators.bash_operator import BashOperator
from airflow.datasets import Dataset
dataset1 = Dataset('s3://folder1/dataset_2.txt')
with DAG(
dag_id="upstream_dag_A",
start_date=datetime(2023, 1, 1),
catchup=False,
schedule="@continuous",
max_active_runs=1,
) as dag:
start_task = BashOperator(
task_id="start_task",
bash_command="echo 'Start task'",
outlets=[dataset1],
)
with DAG(
"downstream-dataset-dag",
start_date=datetime(2023, 1, 1),
schedule=[dataset1],
catchup=False,
max_active_runs=1,
) as dag:
start_task = BashOperator(
task_id="start_task",
# bash_command="echo 'Start task'",
bash_command='sleep 30 && echo "Upstream message: $message"',
)
```
### Operating System
ubuntu
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other 3rd-party Helm chart
### Deployment details
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] 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
Start with the supplied two-DAG reproduction: an @continuous upstream DAG repeatedly updates one Dataset while the downstream DAG sleeps for 30 seconds. Compare the dataset-update and downstream-DAG-run mappings in the UI during multiple updates and after stopping the upstream DAG; done means processed updates are mapped correctly without dangling updates shown as unprocessed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100