This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8a5212c855e Fix Dag run filtering for null start dates (#71654)
8a5212c855e is described below
commit 8a5212c855e8a47468ab62ab82a4d1d2acac1fa1
Author: Vincent Hsiao <[email protected]>
AuthorDate: Wed Oct 21 00:52:38 2026 +0800
Fix Dag run filtering for null start dates (#71654)
* Fix Dag run filtering for null start dates
* Cover Dag list start date filtering with null latest runs
* Preserve queued Dag filtering semantics
* Add unit test for nullable datetime lower bound clause
* Cover nullable datetime strict lower bounds
* Use UTC in nullable datetime filter tests
---
.../airflow/api_fastapi/common/parameters/range.py | 39 +++++++++++++++++-----
.../api_fastapi/core_api/routes/public/dag_run.py | 9 ++++-
.../unit/api_fastapi/common/test_parameters.py | 19 +++++++++++
.../core_api/routes/public/test_dag_run.py | 17 ++++++++++
4 files changed, 74 insertions(+), 10 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/parameters/range.py
b/airflow-core/src/airflow/api_fastapi/common/parameters/range.py
index 9dc587088c6..992e174de52 100644
--- a/airflow-core/src/airflow/api_fastapi/common/parameters/range.py
+++ b/airflow-core/src/airflow/api_fastapi/common/parameters/range.py
@@ -39,7 +39,7 @@ from airflow.typing_compat import Self
if TYPE_CHECKING:
from sqlalchemy.orm.attributes import InstrumentedAttribute
- from sqlalchemy.sql import Select
+ from sqlalchemy.sql import ColumnElement, Select
def _safe_parse_datetime(date_to_check: str) -> datetime:
@@ -136,12 +136,27 @@ class NullableDatetimeRangeFilter(RangeFilter):
``OR`` predicates so each branch can be satisfied by an independent index
scan.
NULL semantics: ``start_date=NULL`` means the task has not started yet;
``end_date=NULL`` means
- the task is still running. For lower bounds the NULL branch passes
unconditionally — a not-yet-
- started/ended task will eventually satisfy any past lower bound. For upper
bounds the NULL branch
- is ``col IS NULL AND now() <= x``, preserving the COALESCE(col, now())
semantics without the
- function-wrap index penalty.
+ the task is still running. For lower bounds the NULL branch usually passes
unconditionally, but
+ callers may narrow it with an extra clause when NULL cannot still evolve
into an in-range value.
+ For upper bounds the NULL branch is ``col IS NULL AND now() <= x``,
preserving the
+ COALESCE(col, now()) semantics without the function-wrap index penalty.
"""
+ def __init__(
+ self,
+ value: Range | None,
+ attribute: InstrumentedAttribute,
+ null_lower_bound_clause: ColumnElement[bool] | None = None,
+ ) -> None:
+ super().__init__(value, attribute)
+ self.null_lower_bound_clause = null_lower_bound_clause
+
+ def _get_null_lower_bound_clause(self) -> ColumnElement[bool]:
+ null_clause: ColumnElement[bool] = self.attribute.is_(None)
+ if self.null_lower_bound_clause is not None:
+ null_clause = and_(null_clause, self.null_lower_bound_clause)
+ return null_clause
+
def to_orm(self, select: Select) -> Select:
if self.skip_none is False:
raise ValueError(f"Cannot set 'skip_none' to False on a
{type(self)}")
@@ -151,10 +166,10 @@ class NullableDatetimeRangeFilter(RangeFilter):
if self.value.lower_bound_gte:
x = self.value.lower_bound_gte
- select = select.where(or_(self.attribute >= x,
self.attribute.is_(None)))
+ select = select.where(or_(self.attribute >= x,
self._get_null_lower_bound_clause()))
if self.value.lower_bound_gt:
x = self.value.lower_bound_gt
- select = select.where(or_(self.attribute > x,
self.attribute.is_(None)))
+ select = select.where(or_(self.attribute > x,
self._get_null_lower_bound_clause()))
if self.value.upper_bound_lte:
x = self.value.upper_bound_lte
select = select.where(or_(self.attribute <= x,
and_(self.attribute.is_(None), func.now() <= x)))
@@ -166,7 +181,11 @@ class NullableDatetimeRangeFilter(RangeFilter):
def datetime_range_filter_factory(
- filter_name: str, model: Base, attribute_name: str | None = None
+ filter_name: str,
+ model: Base,
+ attribute_name: str | None = None,
+ *,
+ null_lower_bound_clause: ColumnElement[bool] | None = None,
) -> Callable[[datetime | None, datetime | None, datetime | None, datetime |
None], RangeFilter]:
def depends_datetime(
lower_bound_gte: datetime | None = Query(alias=f"{filter_name}_gte",
default=None),
@@ -182,7 +201,9 @@ def datetime_range_filter_factory(
upper_bound_lt=upper_bound_lt,
)
if filter_name in ("start_date", "end_date"):
- return NullableDatetimeRangeFilter(range_val, attr)
+ return NullableDatetimeRangeFilter(
+ range_val, attr,
null_lower_bound_clause=null_lower_bound_clause
+ )
return RangeFilter(range_val, attr)
return depends_datetime
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index 68fd621cd3a..bbe47fbde6c 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -518,7 +518,14 @@ def get_dag_runs(
offset: QueryOffset,
run_after: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("run_after", DagRun))],
logical_date: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("logical_date", DagRun))],
- start_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("start_date", DagRun))],
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(
+ datetime_range_filter_factory(
+ "start_date", DagRun,
null_lower_bound_clause=DagRun.end_date.is_(None)
+ )
+ ),
+ ],
end_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("end_date", DagRun))],
duration_range: Annotated[RangeFilter,
Depends(float_range_filter_factory("duration", DagRun))],
update_at_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("updated_at", DagRun))],
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
b/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
index 3640f7646ea..641183a1f32 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_parameters.py
@@ -630,6 +630,25 @@ class TestDatetimeRangeFilterFactory:
assert "now()" not in sql
assert "coalesce" not in sql
+ @pytest.mark.parametrize("lower_bound_name", ["lower_bound_gte",
"lower_bound_gt"])
+ def test_null_lower_bound_clause_is_added_to_null_branch(self,
lower_bound_name):
+ """Custom NULL lower-bound clause is ANDed with the attribute IS NULL
check."""
+ bound = datetime(2026, 5, 3, 12, 0, 0, tzinfo=UTC)
+ depends = datetime_range_filter_factory(
+ "start_date",
+ DagRun,
+ null_lower_bound_clause=DagRun.end_date.is_(None),
+ )
+ bounds = dict(lower_bound_gte=None, lower_bound_gt=None,
upper_bound_lte=None, upper_bound_lt=None)
+ bounds[lower_bound_name] = bound
+ rf = depends(**bounds)
+
+ sql = _compile(rf.to_orm(select(DagRun)))
+
+ assert "start_date is null" in sql
+ assert "end_date is null" in sql
+ assert "start_date is null and" in sql
+
def test_upper_bound_includes_now_for_running_tasks(self):
"""NULL branch on upper bounds uses now() to proxy the in-progress
task's current time."""
bound = datetime(2026, 5, 3, 12, 0, 0, tzinfo=UTC)
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 0fec3f0fe15..c0e3bf9ede8 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -834,6 +834,23 @@ class TestGetDagRuns:
assert len(collected) == len(set(collected)), "cursor pages overlapped"
assert set(collected) == full_ids, "cursor pagination dropped rows
across the NULL boundary"
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def
test_start_date_lower_bound_excludes_finished_runs_with_null_start_date(self,
test_client, session):
+ run = session.scalar(select(DagRun).where(DagRun.run_id ==
DAG1_RUN2_ID))
+ run.start_date = None
+ run.end_date = START_DATE1 + timedelta(minutes=5)
+ session.commit()
+
+ response = test_client.get(
+ "/dags/~/dagRuns",
+ params={"start_date_gte": START_DATE2.isoformat(), "order_by":
"dag_run_id"},
+ )
+
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ assert body["total_entries"] == 2
+ assert [run["dag_run_id"] for run in body["dag_runs"]] ==
[DAG2_RUN1_ID, DAG2_RUN2_ID]
+
@pytest.mark.parametrize(
("dag_id", "query_params", "expected_dag_id_list"),
[
