This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 0807f2b5f75 Reuse the ambient session when skipping pending TIs in
dag-run terminal state (#71488)
0807f2b5f75 is described below
commit 0807f2b5f757a2ab4a908c6194a0f9ac8c9e2f98
Author: Mat <[email protected]>
AuthorDate: Fri Aug 21 17:35:22 2026 +0200
Reuse the ambient session when skipping pending TIs in dag-run terminal
state (#71488)
TaskInstance.set_state is @provide_session: called without a session,
every pending task instance gets its own session — a fresh DB
connection (with the documented sql_alchemy_pool_enabled=False +
pgbouncer setup), a refresh SELECT, a merge SELECT, an UPDATE and a
COMMIT. Marking a dag run failed on a run with 3,917 mapped task
instances took 280 s (~19,500 statements, ~3,900 connections) on an
otherwise idle PostgreSQL 17 with 2.5 ms RTT.
Passing the ambient session through drops the per-TI cost from 79 ms
to 7.5 ms measured on the same run (x10.5); the dag-run state and the
running-TI path already use this session, so transactional semantics
are unchanged.
Co-authored-by: Claude Fable 5 <[email protected]>
---
airflow-core/src/airflow/api/common/mark_tasks.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/api/common/mark_tasks.py
b/airflow-core/src/airflow/api/common/mark_tasks.py
index 9a07e1c9b63..1f597c0ee5f 100644
--- a/airflow-core/src/airflow/api/common/mark_tasks.py
+++ b/airflow-core/src/airflow/api/common/mark_tasks.py
@@ -301,7 +301,7 @@ def _set_dag_run_terminal_state(
if commit:
for ti in pending_normal_tis:
- ti.set_state(TaskInstanceState.SKIPPED)
+ ti.set_state(TaskInstanceState.SKIPPED, session=session)
# Set the dag run state only if there is no pending teardown (else
this would not be scheduled later).
if not any(dag.task_dict[ti.task_id].is_teardown for ti in
(running_tis + pending_tis)):
