This is an automated email from the ASF dual-hosted git repository.
kaxilnaik 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 e796bc23ad3 Optimize task instance API queries by removing redundant
JOINs (#54363)
e796bc23ad3 is described below
commit e796bc23ad373d161bd52ec01f006c1c2cbdf0af
Author: Kaxil Naik <[email protected]>
AuthorDate: Tue Aug 12 14:08:41 2025 +0100
Optimize task instance API queries by removing redundant JOINs (#54363)
* Optimize task instance API queries by removing redundant JOINs
Remove duplicate `.join(TI.dag_run)` calls in task instance API endpoints
that were
redundant with `joinedload(TI.dag_run)` options, resulting in cleaner SQL
execution.
- GET `/dagRuns/{dag_run_id}/taskInstances/{task_id}`
- GET `/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}`
- POST `/dagRuns/~/taskInstances/list`
❌ Before (Redundant):
```python
query = (
select(TI)
.join(TI.dag_run) # Explicit JOIN
.options(joinedload(TI.dag_run)) # Creates ANOTHER JOIN!
)
```
✅ After:
```python
query = (
select(TI)
.options(joinedload(TI.dag_run)) # Single efficient JOIN
)
```
**Before (with redundant JOINs):**
```sql
FROM task_instance
JOIN dag_run ON dag_run.dag_id = task_instance.dag_id AND dag_run.run_id =
task_instance.run_id
JOIN dag_run AS dag_run_1 ON dag_run_1.dag_id = task_instance.dag_id AND
dag_run_1.run_id = task_instance.run_id
LEFT OUTER JOIN dag AS dag_1 ON dag_run_1.dag_id = dag_1.dag_id
```
**After:**
```sql
FROM task_instance
JOIN dag_run AS dag_run_1 ON dag_run_1.dag_id = task_instance.dag_id AND
dag_run_1.run_id = task_instance.run_id
LEFT OUTER JOIN dag AS dag_1 ON dag_run_1.dag_id = dag_1.dag_id
```
Tested using SQLAlchemy query compilation with PostgreSQL to confirm:
- 2 → 1 dag_run JOINs per endpoint
- Maintained all required relationship loading
- No import or syntax errors introduced
* Make test_xcom_get_slice_accepts_include_prior_dates test more resilient
---
.../src/airflow/api_fastapi/core_api/routes/public/task_instances.py | 4 +---
.../tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py | 2 +-
2 files changed, 2 insertions(+), 4 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 f0274d33830..fd5155bb89b 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
@@ -102,7 +102,6 @@ def get_task_instance(
query = (
select(TI)
.where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id ==
task_id)
- .join(TI.dag_run)
.options(joinedload(TI.rendered_task_instance_fields))
.options(joinedload(TI.dag_version))
.options(joinedload(TI.dag_run).options(joinedload(DagRun.dag_model)))
@@ -364,7 +363,6 @@ def get_mapped_task_instance(
query = (
select(TI)
.where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id ==
task_id, TI.map_index == map_index)
- .join(TI.dag_run)
.options(joinedload(TI.rendered_task_instance_fields))
.options(joinedload(TI.dag_version))
.options(joinedload(TI.dag_run).options(joinedload(DagRun.dag_model)))
@@ -544,7 +542,7 @@ def get_task_instances_batch(
TI,
).set_value([body.order_by] if body.order_by else None)
- query = select(TI).join(TI.dag_run)
+ query = select(TI)
task_instance_select, total_entries = paginated_select(
statement=query,
filters=[
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
index 2d03eb45367..0841a76ed1a 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
@@ -320,7 +320,7 @@ class TestXComsGetEndpoint:
)
assert response.status_code == 200
- assert response.json() == expected_xcoms
+ assert set(response.json()) == set(expected_xcoms)
class TestXComsSetEndpoint:
