uranusjr commented on code in PR #70972:
URL: https://github.com/apache/airflow/pull/70972#discussion_r3710779391
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2658,53 +2656,31 @@ def _create_dag_runs_asset_triggered(
)
continue
- triggered_date: DateTime =
timezone.coerce_datetime(queued_adrqs[0].created_at)
- self.log.debug(
- "Creating asset-triggered DagRun for '%s': %d queued assets,
triggered_date=%s",
- dag.dag_id,
- len(queued_adrqs),
- triggered_date,
- )
- cte = (
-
select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
- .where(
- DagRun.dag_id == dag.dag_id,
- DagRun.run_type == DagRunType.ASSET_TRIGGERED,
- DagRun.run_after < triggered_date,
- )
- .cte()
- )
-
- # A first asset-triggered run has no previous run to floor the
event window. With
- # catchup off, floor it at when the Dag started scheduling on its
assets so the
- # backlog is skipped; with catchup on, only date.min applies and
the backlog replays.
- event_window_floor: list[Any] = [cte.c.previous_dag_run_run_after]
- if not dag.catchup:
- event_window_floor.append(
- select(func.min(DagScheduleAssetReference.created_at))
- .where(DagScheduleAssetReference.dag_id == dag.dag_id)
- .scalar_subquery()
+ referenced_event_ids = {adrq.asset_event_id for adrq in
queued_adrqs}
+ event_predicate: ColumnElement[bool] =
AssetEvent.id.in_(referenced_event_ids)
Review Comment:
Unintended, but I would argue this is more correct since the asset event
_was_ used to trigger the run. I think we should keep this behavioral change
with a note somewhere.
--
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]
