diff --git a/airflow-core/src/airflow/models/serialized_dag.py b/airflow-core/src/airflow/models/serialized_dag.py
index e714361df9d8f..401809c5087a7 100644
--- a/airflow-core/src/airflow/models/serialized_dag.py
+++ b/airflow-core/src/airflow/models/serialized_dag.py
@@ -432,7 +432,15 @@ def write_dag(
         has_task_instances: bool = False
         if dag_version:
             has_task_instances = bool(
-                session.scalar(select(exists().where(TaskInstance.dag_version_id == dag_version.id)))
+                session.scalar(
+                    select(
+                        exists().where(
+                            # Using dag_id filter to speed up query via the composite index.
+                            TaskInstance.dag_id == dag.dag_id,
+                            TaskInstance.dag_version_id == dag_version.id,
+                        )
+                    )
+                )
             )
 
         if dag_version and not has_task_instances: