This is an automated email from the ASF dual-hosted git repository.
kaxil 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 38c11037bc1 Take the Postgres migration lock without a bound parameter
(#76976)
38c11037bc1 is described below
commit 38c11037bc11e57f17849629253447a2960fa849
Author: Rahul Vats <[email protected]>
AuthorDate: Thu Nov 12 20:24:52 2026 +0530
Take the Postgres migration lock without a bound parameter (#76976)
The migration lock connection stays idle in transaction for the whole
migration. Under psycopg3 the parameterized `pg_advisory_lock(:id)` keeps
its
snapshot on that idle transaction, so a migration running CREATE INDEX
CONCURRENTLY (e.g. from an external DB manager) waits on its own process's
lock connection and `airflow db migrate` hangs forever.
Inline the lock id instead. The transaction stays open, so behind a
transaction-pooling PgBouncer the lock and unlock still run on the same
server connection.
---
airflow-core/src/airflow/utils/db.py | 16 ++++++++++++----
airflow-core/tests/unit/utils/test_db.py | 22 +++++++++++++++++++++-
2 files changed, 33 insertions(+), 5 deletions(-)
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 5e88b05c137..32291556837 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -1627,7 +1627,12 @@ def _create_global_lock_mysql(lock: DBLocks,
lock_timeout: int) -> Generator[Non
def _create_global_lock_postgresql(
session: Session, lock: DBLocks, lock_timeout: int
) -> Generator[None, None, None]:
- """Create a global advisory lock for PostgreSQL using transactional
advisory locks."""
+ """
+ Create a global advisory lock for PostgreSQL using session-level advisory
locks.
+
+ The lock connection stays in an open transaction while the lock is held,
which keeps it
+ on one server connection behind a transaction-pooling PgBouncer.
+ """
bind = session.get_bind()
if hasattr(bind, "connect"):
conn = bind.connect()
@@ -1642,10 +1647,13 @@ def _create_global_lock_postgresql(
text("SELECT set_config('lock_timeout', :timeout, false)"),
{"timeout": str(lock_timeout)},
)
- conn.execute(text("SELECT pg_advisory_lock(:id)"), {"id":
lock.value})
else:
conn.execute(text("SET LOCK_TIMEOUT to :timeout"), {"timeout":
lock_timeout})
- conn.execute(text("SELECT pg_advisory_lock(:id)"), {"id":
lock.value})
+ # The last statement run on this connection before the yield must have
no bound parameter:
+ # under psycopg3 a parameterized statement keeps its snapshot while
the connection sits idle
+ # in transaction, and CREATE INDEX CONCURRENTLY run by a migration in
this process would
+ # wait on it forever.
+ conn.execute(text(f"SELECT pg_advisory_lock({int(lock.value)})"))
yield
finally:
@@ -1679,7 +1687,7 @@ def create_global_lock(
- GET_LOCK is session-level, not transaction-level
- DDL operations cause implicit commits that would break transaction
wrappers
- For PostgreSQL, uses transactional advisory locks as before.
+ For PostgreSQL, uses a session-level advisory lock, held while the
connection stays in a transaction.
"""
dialect_name = get_dialect_name(session)
diff --git a/airflow-core/tests/unit/utils/test_db.py
b/airflow-core/tests/unit/utils/test_db.py
index 6af67dec69a..786310e3e11 100644
--- a/airflow-core/tests/unit/utils/test_db.py
+++ b/airflow-core/tests/unit/utils/test_db.py
@@ -30,7 +30,7 @@ from alembic.config import Config
from alembic.migration import MigrationContext
from alembic.runtime.environment import EnvironmentContext
from alembic.script import ScriptDirectory
-from sqlalchemy import Column, Integer, MetaData, String, Table, delete, select
+from sqlalchemy import Column, Integer, MetaData, String, Table, delete,
select, text
from sqlalchemy.orm import declarative_base
from airflow import settings
@@ -38,6 +38,7 @@ from airflow.models import Base as airflow_base
from airflow.models.team import Team
from airflow.utils.db import (
AutocommitEngineForMySQL,
+ DBLocks,
LazySelectSequence,
_get_alembic_config,
_get_current_revision,
@@ -46,6 +47,7 @@ from airflow.utils.db import (
check_team_names_can_be_lower_cased,
compare_server_default,
compare_type,
+ create_global_lock,
downgrade,
get_default_connections,
initdb,
@@ -501,6 +503,24 @@ class TestReflectTablesSchemaQualified:
session.execute.assert_called()
[email protected]("postgres")
+def test_postgres_migration_lock_does_not_block_create_index_concurrently():
+ """The held migration lock must not block CREATE INDEX CONCURRENTLY run by
migrations."""
+ engine = settings.get_engine()
+ table = "test_migration_lock_cic"
+ with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as
ddl:
+ ddl.execute(text(f"DROP TABLE IF EXISTS {table}"))
+ ddl.execute(text(f"CREATE TABLE {table} (id INTEGER)"))
+ try:
+ with create_session() as session,
create_global_lock(session=session, lock=DBLocks.MIGRATIONS):
+ # Fails with a statement timeout instead of hanging if the
lock connection blocks it.
+ ddl.execute(text("SET statement_timeout = '10s'"))
+ ddl.execute(text(f"CREATE INDEX CONCURRENTLY idx_{table} ON
{table} (id)"))
+ finally:
+ ddl.execute(text("RESET statement_timeout"))
+ ddl.execute(text(f"DROP TABLE IF EXISTS {table}"))
+
+
class TestAutocommitEngineForMySQL:
"""Test the AutocommitEngineForMySQL context manager."""
