Re: [PR] fix: add temporary index on serialized_dag.dag_id for MySQL in migration 0101 [airflow]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #63797:
URL: https://github.com/apache/airflow/pull/63797#discussion_r3025340427


##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -766,6 +772,9 @@ def _migrate_dag_deadlines(dag_conn: Connection) -> Iterable[str]:
 
         log.info("Batch complete", batch_num=batch_num, total_batches=total_batches)
 
+    if dialect == "mysql":
+        conn.execute(sa.text(f"DROP INDEX {_tmp_idx} ON serialized_dag"))
+

Review Comment:
   `DROP INDEX` is executed only on the happy-path after the batching loop; if anything raises inside the loop, the temporary index will be left behind. That can (1) cause future migration runs to fail on `CREATE INDEX` due to name collision, and (2) leave extra index overhead on `serialized_dag`. Wrap the migration body in `try/finally` to ensure the index is dropped, or better, reuse the existing `temporary_index(...)` helper so cleanup is guaranteed even on error.



##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -806,6 +815,12 @@ def migrate_deadline_alert_data_back_to_serialized_dag() -> None:
     conn = op.get_bind()
     dialect = conn.dialect.name
 
+    # On MySQL, create a temporary index on dag_id to avoid sort buffer overflow
+    # when paginating over large serialized_dag tables (100K+ rows).
+    _tmp_idx = "_tmp_sd_dag_id_idx"
+    if dialect == "mysql":
+        conn.execute(sa.text(f"CREATE INDEX {_tmp_idx} ON serialized_dag (dag_id)"))
+

Review Comment:
   Same issue as in upgrade path: this adds a raw MySQL `CREATE INDEX` that isn’t guaranteed to be dropped on exceptions and will fail on reruns if the index already exists. Please wrap with `try/finally` or use the existing `temporary_index(...)` helper (which checks for an existing matching index and ensures cleanup).



##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -530,6 +530,12 @@ def _migrate_deadline_alerts() -> None:
             ")"
         )
 
+    # On MySQL, create a temporary index on dag_id to avoid sort buffer overflow
+    # when paginating over large serialized_dag tables (100K+ rows).
+    _tmp_idx = "_tmp_sd_dag_id_idx"
+    if dialect == "mysql":
+        conn.execute(sa.text(f"CREATE INDEX {_tmp_idx} ON serialized_dag (dag_id)"))
+

Review Comment:
   Creating `_tmp_sd_dag_id_idx` is risky and likely unnecessary: the pagination queries in this function `ORDER BY id` (PK) and never `ORDER BY dag_id`, so this index won’t address the reported filesort issue but can still fail (e.g., if a prior attempt crashed before dropping the index, `CREATE INDEX` will error on rerun) and can cause heavy DDL/metadata locks on large tables. Consider removing this index entirely, or if an index is truly needed, use the existing `temporary_index(...)` helper (which checks for an existing matching index and guarantees cleanup via `finally`) instead of raw `CREATE INDEX` SQL.
   ```suggestion
   
   ```



##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -530,6 +530,12 @@ def _migrate_deadline_alerts() -> None:
             ")"
         )
 
+    # On MySQL, create a temporary index on dag_id to avoid sort buffer overflow
+    # when paginating over large serialized_dag tables (100K+ rows).
+    _tmp_idx = "_tmp_sd_dag_id_idx"
+    if dialect == "mysql":
+        conn.execute(sa.text(f"CREATE INDEX {_tmp_idx} ON serialized_dag (dag_id)"))

Review Comment:
   Running raw DDL via `conn.execute(sa.text(...))` bypasses Alembic helpers and the file’s existing `temporary_index(...)` utility (which also checks for existing indexes and handles MySQL-specific drop failures). Prefer `op.create_index` / `op.drop_index` (or `temporary_index`) here for consistent migration behavior across backends and to leverage the existing safety logic.
   ```suggestion
           op.create_index(_tmp_idx, "serialized_dag", ["dag_id"])
   ```



##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -919,6 +934,9 @@ def migrate_deadline_alert_data_back_to_serialized_dag() -> None:
 
         log.info("Batch complete", batch_num=batch_num, total_batches=total_batches)
 
+    if dialect == "mysql":
+        conn.execute(sa.text(f"DROP INDEX {_tmp_idx} ON serialized_dag"))
+

Review Comment:
   Same cleanup concern as in upgrade: the index drop is not inside a `finally`, so an exception during processing will leave the index behind and potentially break subsequent migration attempts. Use `try/finally` or the `temporary_index(...)` helper to ensure index removal even on error.



##########
airflow-core/src/airflow/migrations/versions/0101_3_2_0_ui_improvements_for_deadlines.py:
##########
@@ -530,6 +530,12 @@ def _migrate_deadline_alerts() -> None:
             ")"
         )
 
+    # On MySQL, create a temporary index on dag_id to avoid sort buffer overflow
+    # when paginating over large serialized_dag tables (100K+ rows).
+    _tmp_idx = "_tmp_sd_dag_id_idx"

Review Comment:
   The PR description/issue trace references pagination using `ORDER BY dag_id`, but the current migration code paginates with `WHERE id > :last_id` and `ORDER BY id` (including the MySQL subquery). Adding an index on `dag_id` doesn’t align with the actual query shape; either the query should be updated to use `dag_id` ordering (and then the index makes sense) or the description/fix should be adjusted to target the real bottleneck.



-- 
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