This is an automated email from the ASF dual-hosted git repository.
bbovenzi pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new da27a72a271 [v3-4-test] Fix task instance filtering for null start
dates (#75321) (#75324)
da27a72a271 is described below
commit da27a72a271311bed917a980b0ef437c64aa2436
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Oct 21 12:42:31 2026 -0400
[v3-4-test] Fix task instance filtering for null start dates (#75321)
(#75324)
A task instance the executor reports as failed while still queued gets an
end_date but no start_date. The nullable start_date lower bound kept
matching
it as if it had not started yet, so it showed up under any start_date_gte or
start_date_gt filter. Same fix as #71654 for Dag runs.
(cherry picked from commit b99f4a9b3f8067f27560df00ff507743ed52fd7d)
Co-authored-by: Pierre Jeambrun <[email protected]>
---
.../core_api/routes/public/task_instances.py | 14 +++++++++--
.../core_api/routes/public/test_task_instances.py | 28 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 2 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
index 066e4501e36..351f2ec97bb 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py
@@ -214,7 +214,12 @@ def get_mapped_task_instances(
dag_bag: DagBagDep,
run_after_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("run_after", TI))],
logical_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("logical_date", TI))],
- start_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("start_date", TI))],
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(
+ datetime_range_filter_factory("start_date", TI,
null_lower_bound_clause=TI.end_date.is_(None))
+ ),
+ ],
end_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("end_date", TI))],
update_at_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter,
Depends(float_range_filter_factory("duration", TI))],
@@ -560,7 +565,12 @@ def get_task_instances(
task_id: Annotated[FilterParam[str | None],
Depends(filter_param_factory(TI.task_id, str | None))],
run_after_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("run_after", TI))],
logical_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("logical_date", TI))],
- start_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("start_date", TI))],
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(
+ datetime_range_filter_factory("start_date", TI,
null_lower_bound_clause=TI.end_date.is_(None))
+ ),
+ ],
end_date_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("end_date", TI))],
update_at_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter,
Depends(float_range_filter_factory("duration", TI))],
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
index 67d9a6751a3..ddb0776e8b8 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py
@@ -1361,6 +1361,25 @@ class TestGetMappedTaskInstances:
assert body["total_entries"] == 0
assert body["task_instances"] == []
+ def test_start_date_lower_bound_excludes_finished_tis_with_null_start_date(
+ self, test_client, one_task_with_mapped_tis, session
+ ):
+ session.execute(
+ update(TaskInstance)
+ .where(TaskInstance.task_id == "task_2", TaskInstance.map_index ==
0)
+ .values(start_date=None, end_date=DEFAULT_DATETIME_2,
state=TaskInstanceState.FAILED)
+ )
+ session.commit()
+
+ response = test_client.get(
+
"/dags/mapped_tis/dagRuns/run_mapped_tis/taskInstances/task_2/listMapped",
+ params={"start_date_gte": DEFAULT_DATETIME_1},
+ )
+ assert response.status_code == 200
+ body = response.json()
+ assert body["total_entries"] == 2
+ assert [ti["map_index"] for ti in body["task_instances"]] == [1, 2]
+
def test_with_logical_date(self, test_client, one_task_with_mapped_tis):
response = test_client.get(
"/dags/mapped_tis/dagRuns/run_mapped_tis/taskInstances/task_2/listMapped",
@@ -1702,6 +1721,15 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
8,
id="test start_date coalesce with null",
),
+ pytest.param(
+ [{"start_date": None, "end_date": DEFAULT_DATETIME_1}],
+ True,
+
"/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances",
+ {"start_date_gte": DEFAULT_DATETIME_STR_1},
+ 0,
+ 7,
+ id="test start_date lower bound excludes finished with null
start_date",
+ ),
pytest.param(
[
{"pool": "test_pool_1"},
