1fanwang commented on code in PR #68359:
URL: https://github.com/apache/airflow/pull/68359#discussion_r3900607329
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2520,8 +2545,58 @@ def _mark_backfills_complete(self, *, session: Session =
NEW_SESSION) -> None:
for b in backfills:
b.completed_at = now
- def _create_dag_runs(self, dag_models: Collection[DagModel], session:
Session) -> None:
- """Create a DAG run and update the dag_model to control if/when the
next DAGRun should be created."""
+ def _collect_skipped_intervals(
+ self,
+ serdag: SerializedDAG,
+ new_info: DagRunInfo,
+ session: Session,
+ *,
+ listener_has_impls: bool,
+ ) -> SkippedIntervalsSummary | None:
+ """
+ Summarize intervals skipped due to catchup=False.
+
+ Asks the timetable whether the previous automated DagRun's immediate
+ successor (with catchup enabled) is earlier than the new run. Returns
+ ``None`` when there is no schedulable gap or when no previous run
exists.
+ """
+ if serdag.catchup:
+ return None
+ if new_info.data_interval is None:
+ return None
+ if not serdag.has_on_skipped_intervals_callback and not
listener_has_impls:
+ return None
+
+ prev_run = session.scalar(
+ select(DagRun)
+ .where(
+ DagRun.dag_id == serdag.dag_id,
+ DagRun.run_type == DagRunType.SCHEDULED,
+ DagRun.data_interval_end.is_not(None),
+ DagRun.data_interval_end <= new_info.data_interval.start,
Review Comment:
Could this query return the run we just created? `_create_dag_runs()`
[creates `created_run`
first](https://github.com/apache/airflow/blob/518cba794fa7b374a0b6aec2d98449ccf93435f8/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2694-L2725),
then [[looks for the previous
run](https://github.com/apache/airflow/blob/518cba794fa7b374a0b6aec2d98449ccf93435f8/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2570-L2579)](https://github.com/apache/airflow/blob/518cba794fa7b374a0b6aec2d98449ccf93435f8/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2570-L2579).
A `CronTriggerTimetable` run starts and ends at the same time, so the new run
matches `data_interval_end <= new_info.data_interval.start`. SQLAlchemy flushes
pending changes before the query, so it can return the new run as the previous
run and miss the skipped interval. Would it be safer to run this lookup before
`create_dagrun()`, or exclude `created_run`, and add a scheduler test where
`CronTriggerTimetable` s
kips several runs?
--
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]
