This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy 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 4e0f297dc44 Remove redundant ORM result uniquing in core queries
(#69913)
4e0f297dc44 is described below
commit 4e0f297dc44424f903285f3eed3c08647b39cc17
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Wed Jul 15 11:57:54 2026 +0100
Remove redundant ORM result uniquing in core queries (#69913)
The affected queries select unique entities or only eager-load scalar
relationships, so identity filtering adds unnecessary result processing.
---
.../src/airflow/api_fastapi/common/db/dag_runs.py | 2 +-
.../api_fastapi/core_api/routes/public/dag_run.py | 6 ++--
.../core_api/routes/public/task_instances.py | 32 ++++++++--------------
.../api_fastapi/core_api/routes/ui/deadlines.py | 2 +-
.../airflow/api_fastapi/core_api/routes/ui/grid.py | 2 +-
.../core_api/services/public/task_instances.py | 4 +--
.../src/airflow/jobs/scheduler_job_runner.py | 18 +++++-------
7 files changed, 26 insertions(+), 40 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
b/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
index f9ae3d6c3b6..1508ea8528e 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/dag_runs.py
@@ -127,7 +127,7 @@ def attach_dag_versions_to_runs(dag_runs: Sequence[DagRun],
*, session: Session)
.where(DagVersion.id.in_(all_version_ids))
.options(joinedload(DagVersion.bundle))
)
- versions_by_id = {dv.id: dv for dv in
session.scalars(dv_query).unique()}
+ versions_by_id = {dv.id: dv for dv in session.scalars(dv_query)}
versions_per_run: dict[tuple[str, str], dict[UUID, DagVersion]] =
defaultdict(dict)
for row in rows:
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 59b3593b9ed..81d47539229 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
@@ -621,7 +621,7 @@ def get_dag_runs(
dag_run_select, token, order_by,
session.get_bind().dialect.name, is_backward=is_backward
)
- fetched = list(session.scalars(dag_run_select).unique())
+ fetched = list(session.scalars(dag_run_select))
has_more = len(fetched) > page_limit
dag_runs = fetched[:page_limit]
@@ -651,7 +651,7 @@ def get_dag_runs(
limit=limit,
session=session,
)
- dag_runs = list(session.scalars(dag_run_select).unique())
+ dag_runs = list(session.scalars(dag_run_select))
attach_dag_versions_to_runs(dag_runs, session=session)
return DAGRunCollectionResponse(
@@ -933,7 +933,7 @@ def get_list_dag_runs_batch(
session=session,
)
- dag_runs = list(session.scalars(dag_runs_select).unique())
+ dag_runs = list(session.scalars(dag_runs_select))
attach_dag_versions_to_runs(dag_runs, session=session)
return DAGRunCollectionResponse(
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 77f68611c3e..f72cd3801c6 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
@@ -973,16 +973,12 @@ def post_clear_task_instances(
# dag.clear() returns TIs without this relationship loaded; re-query with
joinedload.
# populate_existing=True ensures the joinedload updates TIs already in the
identity map.
if task_instances:
- task_instances = (
- session.scalars(
- select(TI)
- .options(joinedload(TI.rendered_task_instance_fields))
- .where(TI.id.in_([ti.id for ti in task_instances]))
- .execution_options(populate_existing=True)
- )
- .unique()
- .all()
- )
+ task_instances = session.scalars(
+ select(TI)
+ .options(joinedload(TI.rendered_task_instance_fields))
+ .where(TI.id.in_([ti.id for ti in task_instances]))
+ .execution_options(populate_existing=True)
+ ).all()
return TaskInstanceCollectionResponse(
task_instances=[TaskInstanceResponse.model_validate(ti) for ti in
task_instances],
@@ -1139,16 +1135,12 @@ def patch_task_instance_dry_run(
# set_task_instance_state() returns TIs without this relationship loaded;
re-query with joinedload.
# populate_existing=True ensures the joinedload updates TIs already in the
identity map.
if tis:
- tis = (
- session.scalars(
- select(TI)
- .options(joinedload(TI.rendered_task_instance_fields))
- .where(TI.id.in_([ti.id for ti in tis]))
- .execution_options(populate_existing=True)
- )
- .unique()
- .all()
- )
+ tis = session.scalars(
+ select(TI)
+ .options(joinedload(TI.rendered_task_instance_fields))
+ .where(TI.id.in_([ti.id for ti in tis]))
+ .execution_options(populate_existing=True)
+ ).all()
return TaskInstanceCollectionResponse(
task_instances=[
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
index d9cfea10d6d..06eda42ed89 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py
@@ -129,7 +129,7 @@ def get_deadlines(
session=session,
)
- deadlines = session.scalars(deadlines_select).unique()
+ deadlines = session.scalars(deadlines_select)
if dag_run_id != "~" and total_entries == 0:
dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id))
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
index a16f9336fd5..8a0c55d315e 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py
@@ -334,7 +334,7 @@ def get_grid_runs(
limit=limit,
return_total_entries=False,
)
- results = session.execute(dag_runs_select_filter).unique().all()
+ results = session.execute(dag_runs_select_filter).all()
dag_runs = [run for run, _, _ in results]
attach_dag_versions_to_runs(dag_runs, session=session)
grid_runs = []
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py
index 24ae587680a..0a1885f0e26 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py
@@ -140,9 +140,7 @@ def _reload_tis_with_rendered_fields(tis: list[TI],
session: Session) -> list[TI
.options(joinedload(TI.rendered_task_instance_fields))
.where(TI.id.in_([ti.id for ti in tis]))
.execution_options(populate_existing=True)
- )
- .unique()
- .all()
+ ).all()
)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 7d5f7f58020..f6d6a4787f6 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -2417,19 +2417,15 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
# as DagModel.dag_id and DagModel.next_dagrun
# This list is used to verify if the DagRun already exist so that we
don't attempt to create
# duplicate DagRuns
- existing_dagrun_objects = (
- session.scalars(
- select(DagRun)
- .where(
- tuple_(DagRun.dag_id, DagRun.logical_date).in_(
- (dm.dag_id, dm.next_dagrun) for dm in dag_models
- )
+ existing_dagrun_objects = session.scalars(
+ select(DagRun)
+ .where(
+ tuple_(DagRun.dag_id, DagRun.logical_date).in_(
+ (dm.dag_id, dm.next_dagrun) for dm in dag_models
)
- .options(load_only(DagRun.dag_id, DagRun.logical_date))
)
- .unique()
- .all()
- )
+ .options(load_only(DagRun.dag_id, DagRun.logical_date))
+ ).all()
existing_dagruns = {(x.dag_id, x.logical_date): x for x in
existing_dagrun_objects}
# backfill runs are not created by scheduler and their concurrency is
separate
