This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a6b0a7424cd10910d01abf3a9866833a9ea099cc
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 16 17:10:20 2026 +0100
[v3-2-test] [v3-2-test] Ensure that DB migrations run in a single
connection. (#65231) (#65368)
This was discovered by running with a custom External DB manager that had
some
gnarly queries that ended up being locked behind this transaction.
`_single_connection_pool` replaces `settings.engine` with a
SingletonThreadPool engine. But `work_session` was created before that — it
still holds an internal reference to the old engine object.
_single_connection_pool has no way to rebind work_session.
So when _get_current_revision(session=work_session) runs on line 1203 —
inside
the _single_connection_pool() block — it calls session.connection() which
goes
through the old engine, not the SingletonThreadPool. The old engine's pool
was
disposed and recreated empty by engine.dispose(), so it creates a brand new
connection. That connection is completely outside _single_connection_pool's
control.
_single_connection_pool guarantees one connection on the new engine. It
can't
prevent work_session from creating connections on the old one. The name is a
bit of a lie — it's really "single connection pool for new code that uses
settings.engine", not "single connection total."
(cherry picked from commit e3fea3a55564fb4b0daeafe43c521f0ee2001f7a)
(cherry picked from commit f8e0876)
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
airflow-core/src/airflow/utils/db.py | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 9bc0608611b..b848764c558 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -1194,6 +1194,13 @@ def _run_upgradedb(
with _configured_alembic_environment() as env:
source_heads = env.script.get_heads()
+ # End the read-only transaction from _get_current_revision before
+ # external DB manager migrations, which may run DDL that is blocked by
+ # open transactions (e.g. CREATE INDEX CONCURRENTLY). The advisory lock
+ # from create_global_lock() is unaffected: it is session-level and held
+ # on a separate connection.
+ work_session.rollback()
+
if current_revision == source_heads[0] and not
_SKIP_EXTERNAL_DB_MANAGERS_UPGRADE.get():
external_db_manager = RunDBManager()
external_db_manager.upgradedb(work_session,
use_migration_files=use_migration_files)
