jason810496 commented on code in PR #60773:
URL: https://github.com/apache/airflow/pull/60773#discussion_r2938019912
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2016,26 +2016,42 @@ def _create_dag_runs_asset_triggered(
triggered_date_by_dag: dict[str, datetime],
session: Session,
) -> None:
- """For DAGs that are triggered by assets, create dag runs."""
- triggered_dates: dict[str, DateTime] = {
- dag_id: timezone.coerce_datetime(last_asset_event_time)
- for dag_id, last_asset_event_time in triggered_date_by_dag.items()
- }
-
+ """For Dags that are triggered by assets, create Dag runs."""
for dag_model in dag_models:
+ if dag_model.dag_id not in triggered_date_by_dag:
+ continue
Review Comment:
Based on the caller context:
https://github.com/apache/airflow/blob/a89dcaa46dcf7d1cc1a4e2291da87c90f5541bf6/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1823-L1847
It seems the new added `if dag_model.dag_id not in triggered_date_by_dag` is
redundant?
```python
asset_triggered_dags = [d for d in all_dags_needing_dag_runs if d.dag_id in
triggered_date_by_dag]
# ...
self._create_dag_runs_asset_triggered(
dag_models=[d for d in asset_triggered_dags if d.dag_id not in
partition_dag_ids],
```
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2016,26 +2016,42 @@ def _create_dag_runs_asset_triggered(
triggered_date_by_dag: dict[str, datetime],
session: Session,
) -> None:
- """For DAGs that are triggered by assets, create dag runs."""
- triggered_dates: dict[str, DateTime] = {
- dag_id: timezone.coerce_datetime(last_asset_event_time)
- for dag_id, last_asset_event_time in triggered_date_by_dag.items()
- }
-
+ """For Dags that are triggered by assets, create Dag runs."""
for dag_model in dag_models:
+ if dag_model.dag_id not in triggered_date_by_dag:
+ continue
Review Comment:
Now the `triggered_date` will be handled by the last `queued_adrqs`
statement.
So it _seems_ we don't even need to pass `triggered_date_by_dag` to
`_create_dag_runs_asset_triggered` as `triggered_date_by_dag` was only used in
previous version to calcaute `triggered_dates` for each dag_id.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
