You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "manish1337 (via GitHub)" <gi...@apache.org> on 2026/09/29 19:46:44 UTC
[PR] Remove duplicate dag_run (state, dag_id) index on MySQL [airflow]
manish1337 opened a new pull request, #73920:
URL: https://github.com/apache/airflow/pull/73920
`idx_dag_run_queued_dags` and `idx_dag_run_running_dags` are declared on `(state, dag_id)` and only differ by a partial `WHERE` predicate. MySQL has no partial indexes, so on MySQL they are two identical indexes: every `dag_run` write maintains both and the planner has two equivalent choices.
Reproduced with the released images on MySQL 8.0: a fresh `airflow db migrate` on `apache/airflow:3.0.3` and on `3.3.2`, and a 3.0.3 → 3.3.2 upgrade, all end up with both indexes on `(state, dag_id)` — the only exact duplicate in the schema.
**What changes**
- MySQL: the ORM no longer creates `idx_dag_run_queued_dags` (`Index.ddl_if(dialect=("postgresql", "sqlite"))`, so fresh installs are fixed), and a new migration drops it from existing databases. `idx_dag_run_running_dags` is kept because `get_running_dag_runs_to_examine` names it in a MySQL `USE INDEX` hint. The drop/recreate is guarded on `information_schema.STATISTICS`, so databases where someone already dropped it by hand still migrate; offline (`--show-sql-only`) emits the guard as a prepared statement, following migration 0017.
- Postgres: the squashed 2.6.2 migration creates both indexes **without** their `WHERE` predicate, so databases built from migration files (`--use-migration-files`, and Airflow's own test databases) have the same duplicate. The migration recreates them as the partial indexes the ORM declares only when they have no predicate. Databases created by releases (checked: 2.2.5 and 2.3.4 fresh installs upgraded through 2.10.5 to 3.3.2, and a 2.10.5 fresh install) already have the partial indexes and are left untouched.
- SQLite: no change (migration 0017 already fixed the predicates there).
- Alembic autogenerate and `compare_metadata` don't honour `ddl_if`, so `migrations/env.py` hides the index from autogenerate on MySQL and the model/DB sync test ignores it on MySQL. Without the `env.py` change, `alembic check` on MySQL reports `add_index idx_dag_run_queued_dags`.
- New `test_no_duplicate_indexes_in_database` fails if any table has two indexes with the same columns and predicate. It was red on MySQL before this change and caught the Postgres case.
- `test_queued_dags_index_is_created_only_where_partial_indexes_exist` compiles the `dag_run` DDL per dialect, covering the fresh-install (ORM) path that the migration-built test databases don't exercise; it fails on MySQL if the `ddl_if` is removed.
**Checks run locally** (rebased on current `main`; the migration is `0141`, after `e5a91c7f42b3`)
- `breeze run --backend {postgres,mysql,sqlite} pytest airflow-core/tests/unit/utils/test_db.py airflow-core/tests/unit/migrations/test_0141_remove_duplicate_dag_run_state_dag_id_indexes.py --with-db-init`: 34 passed / 34 passed / 33 passed (the rest are backend-specific skips).
- `breeze testing core-tests --skip-db-tests --use-xdist`: 4711 passed.
- `breeze testing core-tests --run-db-tests-only --run-in-parallel` on Postgres 14 and MySQL 8.0 (API, Always, CLI, Core, Other, Serialization): all OK. `breeze testing providers-tests --run-db-tests-only --run-in-parallel` on Postgres 14: all OK. The remaining MySQL/SQLite and provider non-DB runs are still in progress; I'll update here when they finish.
- `airflow db migrate` from this branch (before the rebase) against the MySQL databases created by the 3.0.3 and 3.3.2 images: each ran the new migration (`-> 9f8d3473abf9`) and keeps only `idx_dag_run_running_dags`.
- `airflow db migrate/downgrade --show-sql-only` on MySQL: emits the guarded prepared statements.
- `prek run --from-ref main --stage pre-commit` and `--stage manual` (including `migration-round-trip`): passed.
Out of scope: `idx_dag_run_dag_id (dag_id)` is a left prefix of `dag_id_state` and of the `(dag_id, run_id)` unique key on every backend; that is a separate discussion.
closes: #53509
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Remove duplicate dag_run (state, dag_id) index on MySQL [airflow]
Posted by "manish1337 (via GitHub)" <gi...@apache.org>.
manish1337 commented on code in PR #73920:
URL: https://github.com/apache/airflow/pull/73920#discussion_r4152610894
##########
airflow-core/src/airflow/migrations/versions/0141_3_4_0_remove_duplicate_dag_run_state_dag_id_indexes.py:
##########
@@ -0,0 +1,120 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``.
+
+MySQL has no partial indexes, so ``idx_dag_run_queued_dags`` and ``idx_dag_run_running_dags``
+are both plain indexes on ``(state, dag_id)`` there. ``idx_dag_run_queued_dags`` is dropped;
+``idx_dag_run_running_dags`` is kept because the scheduler names it in a MySQL ``USE INDEX`` hint.
+
+On Postgres the squashed 2.6.2 migration created both without their ``WHERE`` predicate, so
+databases built from migration files (``--use-migration-files``) have the same duplicate.
+Those are recreated as the partial indexes the ORM declares; databases that already have the
+partial indexes are left untouched.
+
+Revision ID: 9f8d3473abf9
+Revises: e5a91c7f42b3
+Create Date: 2026-09-28 12:10:05.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+revision = "9f8d3473abf9"
+down_revision = "e5a91c7f42b3"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+_TABLE = "dag_run"
+_INDEX = "idx_dag_run_queued_dags"
+_EXISTS_QUERY = (
+ "SELECT COUNT(*) FROM information_schema.STATISTICS "
+ f"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{_TABLE}' AND INDEX_NAME = '{_INDEX}'"
+)
+_POSTGRES_PARTIAL_INDEXES = {
+ "idx_dag_run_queued_dags": "queued",
+ "idx_dag_run_running_dags": "running",
+}
+
+
+def _make_postgres_index_partial(conn, index: str, state: str) -> None:
+ """
+ Recreate ``index`` with its ``WHERE state = <state>`` predicate if it currently has none.
+
+ A ``DO`` block keeps the check server-side, so the same SQL works in offline (``--sql``) mode.
+ """
+ conn.execute(
+ sa.text(f"""
+ DO $$
+ BEGIN
+ IF EXISTS (
+ SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('{index}') AND indpred IS NULL
+ ) THEN
+ DROP INDEX {index};
+ CREATE INDEX {index} ON {_TABLE} (state, dag_id) WHERE state = '{state}';
+ END IF;
+ END $$;
+ """)
+ )
+
+
+def _run_if_index_exists(conn, *, exists: bool, statement: str) -> None:
+ """
+ Run ``statement`` only when the index's presence matches ``exists``.
+
+ MySQL has no ``DROP INDEX IF EXISTS`` / ``CREATE INDEX IF NOT EXISTS``, and a user may
+ already have dropped the duplicate by hand. In offline (``--sql``) mode there is no live
+ connection to check against, so the guard is emitted as a prepared statement instead.
+ """
+ if context.is_offline_mode():
+ conn.execute(
+ sa.text(f"""
+ set @var=if(({_EXISTS_QUERY}) {"> 0" if exists else "= 0"},'{statement}','select 1');
+
+ prepare stmt from @var;
+ execute stmt;
+ deallocate prepare stmt;
+ """)
+ )
+ return
+ if bool(conn.execute(sa.text(_EXISTS_QUERY)).scalar()) is exists:
+ conn.execute(sa.text(statement))
+
+
+def upgrade():
+ """Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``."""
+ conn = op.get_bind()
+ if conn.dialect.name == "mysql":
+ _run_if_index_exists(conn, exists=True, statement=f"DROP INDEX {_INDEX} ON {_TABLE}")
+ elif conn.dialect.name == "postgresql":
+ for index, state in _POSTGRES_PARTIAL_INDEXES.items():
+ _make_postgres_index_partial(conn, index, state)
Review Comment:
The squashed 2.6.2 migration creates both of these without the `WHERE` clause ([here](https://github.com/apache/airflow/blob/a65a3ddf8beefc6e7918cb0c9b50ad51ccd2df56/airflow-core/src/airflow/migrations/versions/0000_2_6_2_squashed_migrations.py#L408-L409)), so on Postgres any DB built with `--use-migration-files` (our test DBs included) ends up with the same duplicate as MySQL. This only recreates them when the predicate is missing, so normal installs aren't touched. Happy to drop this part if you'd rather keep the PR MySQL-only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Remove duplicate dag_run (state, dag_id) index on MySQL [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73920:
URL: https://github.com/apache/airflow/pull/73920#discussion_r4154036066
##########
airflow-core/src/airflow/migrations/versions/0142_3_4_0_remove_duplicate_dag_run_state_dag_id_indexes.py:
##########
@@ -0,0 +1,120 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``.
+
+MySQL has no partial indexes, so ``idx_dag_run_queued_dags`` and ``idx_dag_run_running_dags``
+are both plain indexes on ``(state, dag_id)`` there. ``idx_dag_run_queued_dags`` is dropped;
+``idx_dag_run_running_dags`` is kept because the scheduler names it in a MySQL ``USE INDEX`` hint.
+
+On Postgres the squashed 2.6.2 migration created both without their ``WHERE`` predicate, so
+databases built from migration files (``--use-migration-files``) have the same duplicate.
+Those are recreated as the partial indexes the ORM declares; databases that already have the
+partial indexes are left untouched.
+
+Revision ID: 9f8d3473abf9
+Revises: e5a91c7f42b3
+Create Date: 2026-09-28 12:10:05.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+revision = "9f8d3473abf9"
+down_revision = "e5a91c7f42b3"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+_TABLE = "dag_run"
+_INDEX = "idx_dag_run_queued_dags"
+_EXISTS_QUERY = (
+ "SELECT COUNT(*) FROM information_schema.STATISTICS "
+ f"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{_TABLE}' AND INDEX_NAME = '{_INDEX}'"
+)
+_POSTGRES_PARTIAL_INDEXES = {
+ "idx_dag_run_queued_dags": "queued",
+ "idx_dag_run_running_dags": "running",
+}
+
+
+def _make_postgres_index_partial(conn, index: str, state: str) -> None:
+ """
+ Recreate ``index`` with its ``WHERE state = <state>`` predicate if it currently has none.
+
+ A ``DO`` block keeps the check server-side, so the same SQL works in offline (``--sql``) mode.
+ """
+ conn.execute(
+ sa.text(f"""
+ DO $$
+ BEGIN
+ IF EXISTS (
+ SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('{index}') AND indpred IS NULL
+ ) THEN
+ DROP INDEX {index};
+ CREATE INDEX {index} ON {_TABLE} (state, dag_id) WHERE state = '{state}';
+ END IF;
+ END $$;
+ """)
+ )
+
+
+def _run_if_index_exists(conn, *, exists: bool, statement: str) -> None:
+ """
+ Run ``statement`` only when the index's presence matches ``exists``.
+
+ MySQL has no ``DROP INDEX IF EXISTS`` / ``CREATE INDEX IF NOT EXISTS``, and a user may
+ already have dropped the duplicate by hand. In offline (``--sql``) mode there is no live
+ connection to check against, so the guard is emitted as a prepared statement instead.
+ """
+ if context.is_offline_mode():
+ conn.execute(
+ sa.text(f"""
+ set @var=if(({_EXISTS_QUERY}) {"> 0" if exists else "= 0"},'{statement}','select 1');
+
+ prepare stmt from @var;
+ execute stmt;
+ deallocate prepare stmt;
+ """)
+ )
+ return
+ if bool(conn.execute(sa.text(_EXISTS_QUERY)).scalar()) is exists:
+ conn.execute(sa.text(statement))
+
+
+def upgrade():
+ """Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``."""
+ conn = op.get_bind()
+ if conn.dialect.name == "mysql":
+ _run_if_index_exists(conn, exists=True, statement=f"DROP INDEX {_INDEX} ON {_TABLE}")
+ elif conn.dialect.name == "postgresql":
+ for index, state in _POSTGRES_PARTIAL_INDEXES.items():
+ _make_postgres_index_partial(conn, index, state)
Review Comment:
Ohhhh I see. I wasn't even aware of the `--use-migration-files` option.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Remove duplicate dag_run (state, dag_id) index on MySQL [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73920:
URL: https://github.com/apache/airflow/pull/73920#discussion_r4149300880
##########
airflow-core/src/airflow/migrations/versions/0141_3_4_0_remove_duplicate_dag_run_state_dag_id_indexes.py:
##########
@@ -0,0 +1,120 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``.
+
+MySQL has no partial indexes, so ``idx_dag_run_queued_dags`` and ``idx_dag_run_running_dags``
+are both plain indexes on ``(state, dag_id)`` there. ``idx_dag_run_queued_dags`` is dropped;
+``idx_dag_run_running_dags`` is kept because the scheduler names it in a MySQL ``USE INDEX`` hint.
+
+On Postgres the squashed 2.6.2 migration created both without their ``WHERE`` predicate, so
+databases built from migration files (``--use-migration-files``) have the same duplicate.
+Those are recreated as the partial indexes the ORM declares; databases that already have the
+partial indexes are left untouched.
+
+Revision ID: 9f8d3473abf9
+Revises: e5a91c7f42b3
+Create Date: 2026-09-28 12:10:05.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import context, op
+
+revision = "9f8d3473abf9"
+down_revision = "e5a91c7f42b3"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+_TABLE = "dag_run"
+_INDEX = "idx_dag_run_queued_dags"
+_EXISTS_QUERY = (
+ "SELECT COUNT(*) FROM information_schema.STATISTICS "
+ f"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{_TABLE}' AND INDEX_NAME = '{_INDEX}'"
+)
+_POSTGRES_PARTIAL_INDEXES = {
+ "idx_dag_run_queued_dags": "queued",
+ "idx_dag_run_running_dags": "running",
+}
+
+
+def _make_postgres_index_partial(conn, index: str, state: str) -> None:
+ """
+ Recreate ``index`` with its ``WHERE state = <state>`` predicate if it currently has none.
+
+ A ``DO`` block keeps the check server-side, so the same SQL works in offline (``--sql``) mode.
+ """
+ conn.execute(
+ sa.text(f"""
+ DO $$
+ BEGIN
+ IF EXISTS (
+ SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('{index}') AND indpred IS NULL
+ ) THEN
+ DROP INDEX {index};
+ CREATE INDEX {index} ON {_TABLE} (state, dag_id) WHERE state = '{state}';
+ END IF;
+ END $$;
+ """)
+ )
+
+
+def _run_if_index_exists(conn, *, exists: bool, statement: str) -> None:
+ """
+ Run ``statement`` only when the index's presence matches ``exists``.
+
+ MySQL has no ``DROP INDEX IF EXISTS`` / ``CREATE INDEX IF NOT EXISTS``, and a user may
+ already have dropped the duplicate by hand. In offline (``--sql``) mode there is no live
+ connection to check against, so the guard is emitted as a prepared statement instead.
+ """
+ if context.is_offline_mode():
+ conn.execute(
+ sa.text(f"""
+ set @var=if(({_EXISTS_QUERY}) {"> 0" if exists else "= 0"},'{statement}','select 1');
+
+ prepare stmt from @var;
+ execute stmt;
+ deallocate prepare stmt;
+ """)
+ )
+ return
+ if bool(conn.execute(sa.text(_EXISTS_QUERY)).scalar()) is exists:
+ conn.execute(sa.text(statement))
+
+
+def upgrade():
+ """Remove duplicate ``(state, dag_id)`` indexes on ``dag_run``."""
+ conn = op.get_bind()
+ if conn.dialect.name == "mysql":
+ _run_if_index_exists(conn, exists=True, statement=f"DROP INDEX {_INDEX} ON {_TABLE}")
+ elif conn.dialect.name == "postgresql":
+ for index, state in _POSTGRES_PARTIAL_INDEXES.items():
+ _make_postgres_index_partial(conn, index, state)
Review Comment:
What?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org