From 7468b10b587f9a0960da95f41217f123075e9667 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Tue, 3 Feb 2026 15:20:13 -0800 Subject: [PATCH 1/3] Remove align param from iter dagrun infos I don't think this param does anything. It's only set to False when called from within get_run_ids, within set_state, which is only called when marking tasks as failed. At best it seems it is called in an extremely odd and impossible to understand edge case. But let's see what the tests say. --- .../src/airflow/api/common/mark_tasks.py | 2 +- .../airflow/serialization/definitions/dag.py | 22 ------------------- 2 files changed, 1 insertion(+), 23 deletions(-) diff --git a/airflow-core/src/airflow/api/common/mark_tasks.py b/airflow-core/src/airflow/api/common/mark_tasks.py index 09fa9cd5c5e84..95b23cbbc1669 100644 --- a/airflow-core/src/airflow/api/common/mark_tasks.py +++ b/airflow-core/src/airflow/api/common/mark_tasks.py @@ -187,7 +187,7 @@ def get_run_ids(dag: SerializedDAG, run_id: str, future: bool, past: bool, sessi else: dates = [ info.logical_date - for info in dag.iter_dagrun_infos_between(start_date, end_date, align=False) + for info in dag.iter_dagrun_infos_between(start_date, end_date) if info.logical_date # todo: AIP-76 this will not find anything where logical date is null ] run_ids = [dr.run_id for dr in DagRun.find(dag_id=dag.dag_id, logical_date=dates, session=session)] diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py index ea3009c564542..bfdf02bd3ac35 100644 --- a/airflow-core/src/airflow/serialization/definitions/dag.py +++ b/airflow-core/src/airflow/serialization/definitions/dag.py @@ -434,8 +434,6 @@ def iter_dagrun_infos_between( self, earliest: datetime.datetime | None, latest: datetime.datetime, - *, - align: bool = True, ) -> Iterable[DagRunInfo]: """ Yield DagRunInfo using this DAG's timetable between given interval. @@ -443,17 +441,6 @@ def iter_dagrun_infos_between( DagRunInfo instances yielded if their ``logical_date`` is not earlier than ``earliest``, nor later than ``latest``. The instances are ordered by their ``logical_date`` from earliest to latest. - - If ``align`` is ``False``, the first run will happen immediately on - ``earliest``, even if it does not fall on the logical timetable schedule. - The default is ``True``. - - Example: A DAG is scheduled to run every midnight (``0 0 * * *``). If - ``earliest`` is ``2021-06-03 23:00:00``, the first DagRunInfo would be - ``2021-06-03 23:00:00`` if ``align=False``, and ``2021-06-04 00:00:00`` - if ``align=True``. - - # see issue https://github.com/apache/airflow/issues/60455 """ if isinstance(self.timetable, CronPartitionTimetable): # todo: AIP-76 need to update this so that it handles partitions @@ -481,21 +468,12 @@ def iter_dagrun_infos_between( info = None if info is None: - # No runs to be scheduled between the user-supplied timeframe. But - # if align=False, "invent" a data interval for the timeframe itself. - if not align: - yield DagRunInfo.interval(earliest, latest) return if TYPE_CHECKING: # todo: AIP-76 after updating this function for partitions, this may not be true assert info.data_interval is not None - # If align=False and earliest does not fall on the timetable's logical - # schedule, "invent" a data interval for it. - if not align and info.logical_date != earliest: - yield DagRunInfo.interval(earliest, info.data_interval.start) - # Generate naturally according to schedule. while info is not None: yield info From d71086e5c1b210a83f7f054bba93ac01b68dc4e0 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Tue, 3 Feb 2026 17:18:16 -0800 Subject: [PATCH 2/3] update tests --- airflow-core/tests/unit/models/test_dag.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py index 2a45735847ecf..f10eee376c278 100644 --- a/airflow-core/tests/unit/models/test_dag.py +++ b/airflow-core/tests/unit/models/test_dag.py @@ -2746,7 +2746,6 @@ def test_iter_dagrun_infos_between(start_date, expected_infos): iterator = create_scheduler_dag(dag).iter_dagrun_infos_between( earliest=pendulum.instance(start_date), latest=pendulum.instance(DEFAULT_DATE), - align=True, ) assert expected_infos == list(iterator) @@ -2783,7 +2782,7 @@ def _get_registered_timetable(s): ): scheduler_dag = create_scheduler_dag(dag) - iterator = scheduler_dag.iter_dagrun_infos_between(earliest=start, latest=end, align=True) + iterator = scheduler_dag.iter_dagrun_infos_between(earliest=start, latest=end) with caplog.at_level(logging.ERROR): infos = list(iterator) From 333da4f78fa174e52c63bbdae4525ff5aeeb8eb0 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 4 Feb 2026 03:45:46 -0800 Subject: [PATCH 3/3] ensure we include the passed in run --- airflow-core/src/airflow/api/common/mark_tasks.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/api/common/mark_tasks.py b/airflow-core/src/airflow/api/common/mark_tasks.py index 95b23cbbc1669..bb7fee0c2bfa9 100644 --- a/airflow-core/src/airflow/api/common/mark_tasks.py +++ b/airflow-core/src/airflow/api/common/mark_tasks.py @@ -185,11 +185,12 @@ def get_run_ids(dag: SerializedDAG, run_id: str, future: bool, past: bool, sessi elif not dag.timetable.periodic: run_ids = [run_id] else: - dates = [ + dates = {current_logical_date} + dates.update( info.logical_date for info in dag.iter_dagrun_infos_between(start_date, end_date) if info.logical_date # todo: AIP-76 this will not find anything where logical date is null - ] + ) run_ids = [dr.run_id for dr in DagRun.find(dag_id=dag.dag_id, logical_date=dates, session=session)] return run_ids