fix(api): replace COALESCE with index-friendly OR conditions in datetime range filters

[hkc-8010]

Summary

• `datetime_range_filter_factory` wrapped `start_date` and `end_date` columns in
  `COALESCE(column, now())` to treat currently-running tasks (NULL date) as "now".
  This form prevents PostgreSQL from using btree indexes, causing parallel sequential
  scans across the full `task_instance` table even for narrow date windows.
• Replace with explicit OR conditions that are semantically equivalent:
  `(col >= X) OR (col IS NULL AND now() >= X)`
  PostgreSQL can use a btree index on each OR branch via BitmapOr. Running tasks
  (NULL `end_date` / `start_date`) continue to match date-range queries correctly.
• Adds `NullableDatetimeRangeFilter` subclass of `RangeFilter` for the new logic.
  `datetime_range_filter_factory` returns this subclass for `start_date` / `end_date`
  and a plain `RangeFilter` for all other filter names.

Performance

Measured against a 4.6M-row `task_instance` table with an `end_date` btree index:

	Plan	Cost (first 1000 rows)
Before	Parallel Index Scan + Gather Merge (COALESCE wraps column)	19,523
After	Single Index Scan, stops early via LIMIT (OR branches)	2,658

7.4x reduction in query cost.

Note: vanilla Airflow does not ship an `end_date` index by default. The OR form is
the correct architectural fix regardless. Any deployment that adds an `end_date` index
(via a migration) will immediately benefit. A follow-up migration to add the index is
recommended but is outside the scope of this PR.

Changes

• `airflow-core/src/airflow/api_fastapi/common/parameters.py`: add
  `NullableDatetimeRangeFilter`; update `datetime_range_filter_factory`
• `airflow-core/tests/unit/api_fastapi/common/test_parameters.py`: add
  `TestDatetimeRangeFilterFactory` (7 tests covering type dispatch, SQL shape,
  no-COALESCE assertion, and NULL-branch presence)

PR Checklist

• My PR is targeted at the `main` branch
• Tests added (31 pass locally and in Breeze core-tests, Python 3.10 / SQLite)
• `prek` hooks all pass (ruff, ruff-format, mypy)
• No newsfragment needed (internal performance fix, no user-visible behavior change)

closes: #66335

[hkc-8010]

The failed check is CI image checks / Publish documentation and validate versions, and it looks unrelated to this change. It failed while cloning apache/airflow-site with a GitHub-side 500:

fatal: unable to access 'https://github.com/apache/airflow-site.git/': The requested URL returned error: 500

Failing job:
https://github.com/apache/airflow/actions/runs/25672761656/job/75372288729

This looks like transient infra/docs publish fallout rather than a regression from this PR.

[hkc-8010]

Code review

Found 2 issues:

1. Lower-bound NULL branch still includes func.now() >= x, which breaks filter semantics for future-scheduled tasks. For a lower bound, a task with NULL start_date (not yet started) should unconditionally pass — its eventual value will always be >= any past bound. With now() >= x, if x is in the future, now() >= x is FALSE, so future-scheduled tasks are incorrectly excluded. Reviewer ashb explicitly called this out and suggested or_(self.attribute.is_(None), self.attribute >= x) for lower bounds. The author's reply acknowledges this, but the code and the test test_lower_bound_calls_now_for_running_tasks (which asserts "now()" still appears in lower-bound SQL) contradict the claimed fix.

airflow/airflow-core/src/airflow/api_fastapi/common/parameters.py

Lines 907 to 916 in 315d94d

	ifself.value.lower_bound_gte:
	x=self.value.lower_bound_gte
	select=select.where(or_(self.attribute>=x, and_(self.attribute.is_(None), func.now() >=x)))
	ifself.value.lower_bound_gt:
	x=self.value.lower_bound_gt
	select=select.where(or_(self.attribute>x, and_(self.attribute.is_(None), func.now() >x)))
	ifself.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)))

2. The factory dispatch if filter_name in ("start_date", "end_date") checks the API-facing filter name, not the resolved attribute name. Callers in dags.py pass aliased names ("dag_run_start_date", "dag_run_end_date") with attribute_name="start_date" / "end_date", so they fall through to a plain RangeFilter. Since DagRun.end_date is NULL for running DAGs, these callers silently miss the index-friendly optimization and the correct NULL semantics. The test test_attribute_name_alias_returns_plain_range_filter asserts this as correct behavior, but that contradicts the PR's stated goal. The fix from a previous iteration of this PR (attribute_name or filter_name) would cover these callers.

airflow/airflow-core/src/airflow/api_fastapi/common/parameters.py

Lines 938 to 945 in 315d94d

	upper_bound_lt=upper_bound_lt,
	)
	iffilter_namein ("start_date", "end_date"):
	returnNullableDatetimeRangeFilter(range_val, attr)
	returnRangeFilter(range_val, attr)
	returndepends_datetime

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[hkc-8010]

Both issues from the code review above have been addressed in the latest push (commit 2ff659c):

Issue 1 — lower-bound now() guard removed

Lower bounds now use or_(col >= x, col.is_(None)) with no func.now() call. A task with NULL start_date passes any lower-bound filter unconditionally, which is the correct semantic: a not-yet-started task will eventually start at or after any past bound.

Issue 2 — factory dispatch fixed for aliased filter names

The dispatch condition changed from filter_name in ("start_date", "end_date") to (attribute_name or filter_name) in ("start_date", "end_date"). Callers that pass an aliased filter_name with attribute_name="start_date" or attribute_name="end_date" (e.g. the DAGs endpoint using "dag_run_end_date") now correctly receive NullableDatetimeRangeFilter rather than a plain RangeFilter.

[potiuk]

@hkc-8010 — There is 1 unresolved review thread on this PR from @ashb, and you have engaged with each one (post-review commits and/or in-thread replies). Could you confirm whether you believe the feedback is fully addressed and the PR is ready for maintainer review confirmation?

If yes, reply here (a short "yes / ready" is fine) and an Apache Airflow maintainer will pick the PR up from the review queue on the next sweep.

If you are still working on a thread, please reply with what is outstanding so the threads stay unresolved on purpose.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[hkc-8010]

The PR is ready for review.

[github-actions]

Backport successfully created: v3-2-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-2-test