You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Lee-W (via GitHub)" <gi...@apache.org> on 2026/04/20 14:58:45 UTC
[PR] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Lee-W opened a new pull request, #65547:
URL: https://github.com/apache/airflow/pull/65547
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] 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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #65547:
URL: https://github.com/apache/airflow/pull/65547#issuecomment-4333882390
cc @Dev-iL
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W merged PR #65547:
URL: https://github.com/apache/airflow/pull/65547
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #65547:
URL: https://github.com/apache/airflow/pull/65547#issuecomment-4377920820
### Backport failed to create: v3-2-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/25367646741'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-2-test</td>
<td><a href="https://github.com/apache/airflow/commit/1467d5b4bd95de06cb5340a2c8e64abeb0a03e06"><img src='https://img.shields.io/badge/Commit-1467d5b-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker 1467d5b v3-2-test
```
This should apply the commit to the v3-2-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#how-to-backport-pr-with-cherry-picker-cli).
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #65547:
URL: https://github.com/apache/airflow/pull/65547#discussion_r3167330001
##########
scripts/in_container/run_migration_round_trip.py:
##########
@@ -0,0 +1,336 @@
+#!/usr/bin/env python
+# 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.
+"""
+Migration round-trip regression check — intended to run inside Breeze.
+
+Walks every revision base → head (upgrade direction) and head → base
+(downgrade direction) one rev at a time, on a single fresh SQLite DB
+with ``PRAGMA foreign_keys = ON``. Before each step the seed fixture is
+idempotently restored so any FK chain a rebuild needs to fire is in
+place.
+
+Per-migration walking is essential. ``airflow_db.upgradedb`` /
+``airflow_db.downgrade`` reconfigures Airflow's engine via
+``_single_connection_pool`` for every call, so each migration starts
+on a freshly-connected SQLAlchemy engine where ``setup_event_handlers``
+fires ``PRAGMA foreign_keys=ON``. In contrast a single-call
+``upgradedb(to_revision=tip)`` walks all migrations on one shared
+connection: once an earlier migration's ``disable_sqlite_fkeys`` flips
+FK off on that connection, the off state simply persists into later
+migrations, silently masking broken or missing wrappers downstream.
+
+The placement convention every migration must follow (only DML breaks
+PRAGMA, but the safe rule is strictly stronger) is documented in
+``contributing-docs/26_migration_round_trip_check.rst``.
+
+For the placement convention every migration must follow and the FK
+chains exercised by the seed, see
+``contributing-docs/26_migration_round_trip_check.rst``.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from alembic.script import ScriptDirectory
+from rich.console import Console
+from rich.panel import Panel
+from rich.rule import Rule
+from sqlalchemy import text
+
+console = Console(width=140, color_system="standard", highlight=False)
+
+SEED_DAG_ID = "round_trip_dag"
+SEED_DAG_VERSION_ID = "abcdef0123456789abcdef0123456789"
+SEED_BUNDLE_NAME = "dags-folder"
+SEED_RUN_ID = "manual_run_1"
+SEED_TASK_ID = "task1"
+SEED_TASK_INSTANCE_ID = "fedcba9876543210fedcba9876543210"
+
+# Hardcoded INSERT values for tables we seed. Each entry maps column name
+# to a SQL literal. When a column is not present in the live schema (e.g.
+# an older revision predating its addition), it is dropped from the INSERT
+# — provided every NOT-NULL-no-default column at that revision is covered.
+# When a future migration introduces a brand-new NOT NULL column without a
+# default, ``_insert_or_skip`` raises with a clear message pointing at the
+# table that needs an entry here.
+SEED_VALUES: dict[str, dict[str, str]] = {
+ "log_template": {
+ "id": "1",
+ "filename": "'{{ ti.dag_id }}/{{ ts }}.log'",
+ "elasticsearch_id": "'{{ ti.dag_id }}-{{ ts }}'",
+ "created_at": "'2024-01-01 00:00:00.000000+00:00'",
+ },
+ "dag_bundle": {
+ "name": f"'{SEED_BUNDLE_NAME}'",
+ "active": "1",
+ "version": "NULL",
+ "last_refreshed": "NULL",
+ "signed_url_template": "NULL",
+ "url_template": "NULL",
+ "bundle_template_params": "NULL",
+ },
+ "dag": {
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "max_active_tasks": "16",
+ "has_task_concurrency_limits": "0",
+ "is_paused": "0",
+ "is_active": "1",
+ "is_stale": "0",
+ "fileloc": "'/tmp/round_trip_dag.py'",
+ "fail_fast": "0",
+ "max_consecutive_failed_dag_runs": "0",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "has_import_errors": "0",
+ "max_active_runs": "16",
+ "_default_view": "'grid'",
+ "exceeds_max_non_backfill": "0",
+ "timetable_type": "'cron'",
+ "timetable_partitioned": "0",
+ "timetable_periodic": "0",
+ },
+ "dag_version": {
+ "id": f"'{SEED_DAG_VERSION_ID}'",
+ "version_number": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "bundle_version": "NULL",
+ "created_at": "'2024-01-01 00:00:00+00:00'",
+ "last_updated": "'2024-01-01 00:00:00+00:00'",
+ },
+ "dag_run": {
+ "id": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "execution_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "logical_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "run_type": "'manual'",
+ "state": "'success'",
+ "log_template_id": "1",
+ "run_after": "'2024-01-01 00:00:00+00:00'",
+ "span_status": "'not_started'",
+ "clear_number": "0",
+ },
+ "task_instance": {
+ "id": f"'{SEED_TASK_INSTANCE_ID}'",
+ "task_id": f"'{SEED_TASK_ID}'",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "map_index": "-1",
+ "pool": "'default'",
+ "pool_slots": "1",
+ "state": "'success'",
+ "try_number": "1",
+ "max_tries": "0",
+ },
+}
+
+# FK-dependency-respecting insertion order.
+SEED_ORDER = ("log_template", "dag_bundle", "dag", "dag_version", "dag_run", "task_instance")
+
+
+def _make_fresh_db() -> Path:
+ """Create a fresh SQLite file and re-point Airflow's engine at it."""
+ fd, path = tempfile.mkstemp(prefix="round_trip_", suffix=".db")
+ os.close(fd)
+ db = Path(path)
+ db.unlink() # Airflow creates the file on first connect
+
+ new_conn = f"sqlite:///{db}"
+ os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = new_conn
+
+ from airflow import settings
+
+ settings.SQL_ALCHEMY_CONN = new_conn
+ settings.dispose_orm()
+ settings.configure_orm()
+ return db
Review Comment:
it's intentional
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #65547:
URL: https://github.com/apache/airflow/pull/65547#issuecomment-4377906119
haha, it's not as frequent. so this should be enough
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #65547:
URL: https://github.com/apache/airflow/pull/65547#discussion_r3154020811
##########
scripts/in_container/run_migration_round_trip.py:
##########
@@ -0,0 +1,336 @@
+#!/usr/bin/env python
+# 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.
+"""
+Migration round-trip regression check — intended to run inside Breeze.
+
+Walks every revision base → head (upgrade direction) and head → base
+(downgrade direction) one rev at a time, on a single fresh SQLite DB
+with ``PRAGMA foreign_keys = ON``. Before each step the seed fixture is
+idempotently restored so any FK chain a rebuild needs to fire is in
+place.
+
+Per-migration walking is essential. ``airflow_db.upgradedb`` /
+``airflow_db.downgrade`` reconfigures Airflow's engine via
+``_single_connection_pool`` for every call, so each migration starts
+on a freshly-connected SQLAlchemy engine where ``setup_event_handlers``
+fires ``PRAGMA foreign_keys=ON``. In contrast a single-call
+``upgradedb(to_revision=tip)`` walks all migrations on one shared
+connection: once an earlier migration's ``disable_sqlite_fkeys`` flips
+FK off on that connection, the off state simply persists into later
+migrations, silently masking broken or missing wrappers downstream.
+
+The placement convention every migration must follow (only DML breaks
+PRAGMA, but the safe rule is strictly stronger) is documented in
+``contributing-docs/26_migration_round_trip_check.rst``.
+
+For the placement convention every migration must follow and the FK
+chains exercised by the seed, see
+``contributing-docs/26_migration_round_trip_check.rst``.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from alembic.script import ScriptDirectory
+from rich.console import Console
+from rich.panel import Panel
+from rich.rule import Rule
+from sqlalchemy import text
+
+console = Console(width=140, color_system="standard", highlight=False)
+
+SEED_DAG_ID = "round_trip_dag"
+SEED_DAG_VERSION_ID = "abcdef0123456789abcdef0123456789"
+SEED_BUNDLE_NAME = "dags-folder"
+SEED_RUN_ID = "manual_run_1"
+SEED_TASK_ID = "task1"
+SEED_TASK_INSTANCE_ID = "fedcba9876543210fedcba9876543210"
+
+# Hardcoded INSERT values for tables we seed. Each entry maps column name
+# to a SQL literal. When a column is not present in the live schema (e.g.
+# an older revision predating its addition), it is dropped from the INSERT
+# — provided every NOT-NULL-no-default column at that revision is covered.
+# When a future migration introduces a brand-new NOT NULL column without a
+# default, ``_insert_or_skip`` raises with a clear message pointing at the
+# table that needs an entry here.
+SEED_VALUES: dict[str, dict[str, str]] = {
+ "log_template": {
+ "id": "1",
+ "filename": "'{{ ti.dag_id }}/{{ ts }}.log'",
+ "elasticsearch_id": "'{{ ti.dag_id }}-{{ ts }}'",
+ "created_at": "'2024-01-01 00:00:00.000000+00:00'",
+ },
+ "dag_bundle": {
+ "name": f"'{SEED_BUNDLE_NAME}'",
+ "active": "1",
+ "version": "NULL",
+ "last_refreshed": "NULL",
+ "signed_url_template": "NULL",
+ "url_template": "NULL",
+ "bundle_template_params": "NULL",
+ },
+ "dag": {
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "max_active_tasks": "16",
+ "has_task_concurrency_limits": "0",
+ "is_paused": "0",
+ "is_active": "1",
+ "is_stale": "0",
+ "fileloc": "'/tmp/round_trip_dag.py'",
+ "fail_fast": "0",
+ "max_consecutive_failed_dag_runs": "0",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "has_import_errors": "0",
+ "max_active_runs": "16",
+ "_default_view": "'grid'",
+ "exceeds_max_non_backfill": "0",
+ "timetable_type": "'cron'",
+ "timetable_partitioned": "0",
+ "timetable_periodic": "0",
+ },
+ "dag_version": {
+ "id": f"'{SEED_DAG_VERSION_ID}'",
+ "version_number": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "bundle_version": "NULL",
+ "created_at": "'2024-01-01 00:00:00+00:00'",
+ "last_updated": "'2024-01-01 00:00:00+00:00'",
+ },
+ "dag_run": {
+ "id": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "execution_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "logical_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "run_type": "'manual'",
+ "state": "'success'",
+ "log_template_id": "1",
+ "run_after": "'2024-01-01 00:00:00+00:00'",
+ "span_status": "'not_started'",
+ "clear_number": "0",
+ },
+ "task_instance": {
+ "id": f"'{SEED_TASK_INSTANCE_ID}'",
+ "task_id": f"'{SEED_TASK_ID}'",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "map_index": "-1",
+ "pool": "'default'",
+ "pool_slots": "1",
+ "state": "'success'",
+ "try_number": "1",
+ "max_tries": "0",
+ },
+}
+
+# FK-dependency-respecting insertion order.
+SEED_ORDER = ("log_template", "dag_bundle", "dag", "dag_version", "dag_run", "task_instance")
+
+
+def _make_fresh_db() -> Path:
+ """Create a fresh SQLite file and re-point Airflow's engine at it."""
+ fd, path = tempfile.mkstemp(prefix="round_trip_", suffix=".db")
+ os.close(fd)
+ db = Path(path)
+ db.unlink() # Airflow creates the file on first connect
+
+ new_conn = f"sqlite:///{db}"
+ os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = new_conn
+
+ from airflow import settings
+
+ settings.SQL_ALCHEMY_CONN = new_conn
+ settings.dispose_orm()
+ settings.configure_orm()
+ return db
+
+
+def _step(msg: str) -> None:
+ console.print()
+ console.print(Rule(f"[bold cyan]{msg}[/]", style="cyan"))
+
+
+def _table_info(conn, table: str) -> list[tuple[Any, ...]]:
+ return list(conn.execute(text(f"PRAGMA table_info('{table}')")).fetchall())
+
+
+def _table_exists(conn, table: str) -> bool:
+ return bool(
+ conn.execute(
+ text("SELECT 1 FROM sqlite_master WHERE type='table' AND name=:n"),
+ {"n": table},
+ ).scalar()
+ )
+
+
+def _insert_or_skip(conn, table: str) -> None:
+ """Insert a seed row into ``table`` if the table exists and the row is missing.
+
+ Only columns present in the live schema and listed in ``SEED_VALUES`` are
+ included. Raises if a NOT-NULL-no-default column at the live revision is
+ not in ``SEED_VALUES``.
+ """
+ if not _table_exists(conn, table):
+ return
+
+ info = _table_info(conn, table)
+ live_cols = {r[1] for r in info}
+ not_null_no_default = {r[1] for r in info if r[3] and r[4] is None}
+
+ seed = SEED_VALUES[table]
+ use = {k: v for k, v in seed.items() if k in live_cols}
+ missing = not_null_no_default - set(use)
+ if missing:
+ raise RuntimeError(
+ f"Cannot seed `{table}`: NOT-NULL columns without defaults are missing seed values: "
+ f"{sorted(missing)}. Add them to SEED_VALUES['{table}'] in this script."
+ )
+
+ key = "dag_id" if "dag_id" in use else ("id" if "id" in use else next(iter(use)))
+ if conn.execute(text(f"SELECT 1 FROM {table} WHERE {key}={use[key]}")).scalar():
+ return
+
+ cols = ", ".join(use)
+ vals = ", ".join(use.values())
+ conn.execute(text(f"INSERT INTO {table} ({cols}) VALUES ({vals})"))
+
+
+def _seed_for_current_schema() -> None:
+ """Idempotently seed every table the live schema can accept.
+
+ Run before each migration step so the seed is in place to fire FK-cascade
+ chains during the rebuild that step performs (if any).
+ """
+ from airflow import settings
+
+ with settings.get_engine().begin() as conn:
+ for table in SEED_ORDER:
+ _insert_or_skip(conn, table)
+
+ ti_cols = (
+ {r[1] for r in _table_info(conn, "task_instance")}
+ if _table_exists(conn, "task_instance")
+ else set()
+ )
+ if "dag_version_id" in ti_cols and _table_exists(conn, "dag_version"):
+ conn.execute(
+ text(
+ "UPDATE task_instance SET dag_version_id = :dvid "
+ "WHERE task_id = :tid AND dag_id = :did AND dag_version_id IS NULL"
+ ),
+ {"dvid": SEED_DAG_VERSION_ID, "tid": SEED_TASK_ID, "did": SEED_DAG_ID},
+ )
+
+
+def _all_revisions() -> list[str]:
+ """Return every migration revision in topological order, base → head."""
+ from airflow.utils.db import _get_alembic_config
+
+ cfg = _get_alembic_config()
+ script = ScriptDirectory.from_config(cfg)
+ return [r.revision for r in reversed(list(script.walk_revisions(base="base", head="heads")))]
+
+
+def main() -> None:
+ revs = _all_revisions()
+ console.print(
+ Panel(
+ f"Walking {len(revs)} revisions: base → head, then head → base.",
+ title="migration round-trip",
+ border_style="cyan",
+ expand=False,
+ )
+ )
+
+ db = _make_fresh_db()
+ try:
+ from airflow.utils import db as airflow_db
+ from airflow.utils.db import _SKIP_EXTERNAL_DB_MANAGERS_UPGRADE
+
+ _SKIP_EXTERNAL_DB_MANAGERS_UPGRADE.set(True)
+
+ _step(f"Walk UP base → {revs[-1][:8]}…")
+ for rev in revs:
+ try:
+ _seed_for_current_schema()
+ except Exception as exc:
+ _fail("seed", rev, exc)
+ return
+ try:
+ airflow_db.upgradedb(to_revision=rev)
+ except Exception as exc:
+ _fail("up", rev, exc)
+ return
+
+ _step(f"Walk DOWN {revs[-1][:8]}… → {revs[0][:8]}…")
+ # Stop at the first migration (the squashed-migrations file). Going
+ # all the way to ``base`` means executing the squashed migration's
+ # downgrade, which "drop everything we created" can fail on objects
+ # that later migrations have already renamed or removed (e.g. the
+ # ``sm_dag`` index). That's a pre-existing limitation of the
+ # squashed-migrations file, not a FK round-trip concern.
+ down_revs = list(reversed(revs[1:]))
+ down_targets = list(reversed(revs[:-1]))
+ for downgrading_rev, target in zip(down_revs, down_targets):
+ try:
+ _seed_for_current_schema()
+ except Exception as exc:
+ _fail("seed", downgrading_rev, exc)
+ return
+ try:
+ airflow_db.downgrade(to_revision=target)
+ except Exception as exc:
+ _fail("down", downgrading_rev, exc)
+ return
+
+ console.print()
+ console.print(
+ Panel(
+ f"[bold green]PASS[/] — {len(revs)} revisions round-tripped clean",
+ border_style="green",
+ expand=False,
+ )
+ )
Review Comment:
`_SKIP_EXTERNAL_DB_MANAGERS_UPGRADE` is set but never reset. Even though this is a short-lived script, using the ContextVar token/reset pattern (like `initdb()` does) avoids leaking this state into any later calls within the same process and makes the intent clearer.
```suggestion
skip_external_db_managers_upgrade_token = _SKIP_EXTERNAL_DB_MANAGERS_UPGRADE.set(True)
try:
_step(f"Walk UP base → {revs[-1][:8]}…")
for rev in revs:
try:
_seed_for_current_schema()
except Exception as exc:
_fail("seed", rev, exc)
return
try:
airflow_db.upgradedb(to_revision=rev)
except Exception as exc:
_fail("up", rev, exc)
return
_step(f"Walk DOWN {revs[-1][:8]}… → {revs[0][:8]}…")
# Stop at the first migration (the squashed-migrations file). Going
# all the way to ``base`` means executing the squashed migration's
# downgrade, which "drop everything we created" can fail on objects
# that later migrations have already renamed or removed (e.g. the
# ``sm_dag`` index). That's a pre-existing limitation of the
# squashed-migrations file, not a FK round-trip concern.
down_revs = list(reversed(revs[1:]))
down_targets = list(reversed(revs[:-1]))
for downgrading_rev, target in zip(down_revs, down_targets):
try:
_seed_for_current_schema()
except Exception as exc:
_fail("seed", downgrading_rev, exc)
return
try:
airflow_db.downgrade(to_revision=target)
except Exception as exc:
_fail("down", downgrading_rev, exc)
return
console.print()
console.print(
Panel(
f"[bold green]PASS[/] — {len(revs)} revisions round-tripped clean",
border_style="green",
expand=False,
)
)
finally:
_SKIP_EXTERNAL_DB_MANAGERS_UPGRADE.reset(skip_external_db_managers_upgrade_token)
```
##########
contributing-docs/14_metadata_database_updates.rst:
##########
@@ -97,6 +97,18 @@ and back down to the former. To run any of those CI tests on your machine, you c
1. Copy the relevant command (specified by the ``run`` key for the relevant CI job), and replace the environment variable references with their literal values defined in the sibling ``env`` section.
2. Run the command you created from step 1, troubleshooting errors as needed.
+SQLite FK round-trip safety
+---------------------------
+
+Migrations that rebuild a parent table via ``op.batch_alter_table`` must
+wrap their entire body in ``disable_sqlite_fkeys(op)`` *before* any DML or
+DDL opens an implicit transaction — otherwise the wrapper's PRAGMA is a
+no-op and the rebuild's implicit ``DROP TABLE`` cascade-deletes child rows
+(or aborts on a RESTRICT chain). The placement convention, the
+per-airflow-version round-trip prek hook that enforces it, and the
+baseline file used to tolerate already-known placement bugs are documented
+in `Migration round-trip regression check <26_migration_round_trip_check.rst>`__.
Review Comment:
`disable_sqlite_fkeys` guidance mentions a “baseline file used to tolerate already-known placement bugs”, but there doesn’t appear to be any such baseline file added/referenced elsewhere in the repo. This reads like stale/incomplete documentation—either add the referenced baseline mechanism/file or remove this clause so the section only describes what actually exists.
```suggestion
(or aborts on a RESTRICT chain). The placement convention and the
per-airflow-version round-trip pre-commit hook that enforces it are
documented in `Migration round-trip regression check <26_migration_round_trip_check.rst>`__.
```
##########
scripts/in_container/run_migration_round_trip.py:
##########
@@ -0,0 +1,336 @@
+#!/usr/bin/env python
+# 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.
+"""
+Migration round-trip regression check — intended to run inside Breeze.
+
+Walks every revision base → head (upgrade direction) and head → base
+(downgrade direction) one rev at a time, on a single fresh SQLite DB
+with ``PRAGMA foreign_keys = ON``. Before each step the seed fixture is
+idempotently restored so any FK chain a rebuild needs to fire is in
+place.
+
+Per-migration walking is essential. ``airflow_db.upgradedb`` /
+``airflow_db.downgrade`` reconfigures Airflow's engine via
+``_single_connection_pool`` for every call, so each migration starts
+on a freshly-connected SQLAlchemy engine where ``setup_event_handlers``
+fires ``PRAGMA foreign_keys=ON``. In contrast a single-call
+``upgradedb(to_revision=tip)`` walks all migrations on one shared
+connection: once an earlier migration's ``disable_sqlite_fkeys`` flips
+FK off on that connection, the off state simply persists into later
+migrations, silently masking broken or missing wrappers downstream.
+
+The placement convention every migration must follow (only DML breaks
+PRAGMA, but the safe rule is strictly stronger) is documented in
+``contributing-docs/26_migration_round_trip_check.rst``.
+
+For the placement convention every migration must follow and the FK
+chains exercised by the seed, see
+``contributing-docs/26_migration_round_trip_check.rst``.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from alembic.script import ScriptDirectory
+from rich.console import Console
+from rich.panel import Panel
+from rich.rule import Rule
+from sqlalchemy import text
+
+console = Console(width=140, color_system="standard", highlight=False)
+
+SEED_DAG_ID = "round_trip_dag"
+SEED_DAG_VERSION_ID = "abcdef0123456789abcdef0123456789"
+SEED_BUNDLE_NAME = "dags-folder"
+SEED_RUN_ID = "manual_run_1"
+SEED_TASK_ID = "task1"
+SEED_TASK_INSTANCE_ID = "fedcba9876543210fedcba9876543210"
+
+# Hardcoded INSERT values for tables we seed. Each entry maps column name
+# to a SQL literal. When a column is not present in the live schema (e.g.
+# an older revision predating its addition), it is dropped from the INSERT
+# — provided every NOT-NULL-no-default column at that revision is covered.
+# When a future migration introduces a brand-new NOT NULL column without a
+# default, ``_insert_or_skip`` raises with a clear message pointing at the
+# table that needs an entry here.
+SEED_VALUES: dict[str, dict[str, str]] = {
+ "log_template": {
+ "id": "1",
+ "filename": "'{{ ti.dag_id }}/{{ ts }}.log'",
+ "elasticsearch_id": "'{{ ti.dag_id }}-{{ ts }}'",
+ "created_at": "'2024-01-01 00:00:00.000000+00:00'",
+ },
+ "dag_bundle": {
+ "name": f"'{SEED_BUNDLE_NAME}'",
+ "active": "1",
+ "version": "NULL",
+ "last_refreshed": "NULL",
+ "signed_url_template": "NULL",
+ "url_template": "NULL",
+ "bundle_template_params": "NULL",
+ },
+ "dag": {
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "max_active_tasks": "16",
+ "has_task_concurrency_limits": "0",
+ "is_paused": "0",
+ "is_active": "1",
+ "is_stale": "0",
+ "fileloc": "'/tmp/round_trip_dag.py'",
+ "fail_fast": "0",
+ "max_consecutive_failed_dag_runs": "0",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "has_import_errors": "0",
+ "max_active_runs": "16",
+ "_default_view": "'grid'",
+ "exceeds_max_non_backfill": "0",
+ "timetable_type": "'cron'",
+ "timetable_partitioned": "0",
+ "timetable_periodic": "0",
+ },
+ "dag_version": {
+ "id": f"'{SEED_DAG_VERSION_ID}'",
+ "version_number": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "bundle_version": "NULL",
+ "created_at": "'2024-01-01 00:00:00+00:00'",
+ "last_updated": "'2024-01-01 00:00:00+00:00'",
+ },
+ "dag_run": {
+ "id": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "execution_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "logical_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "run_type": "'manual'",
+ "state": "'success'",
+ "log_template_id": "1",
+ "run_after": "'2024-01-01 00:00:00+00:00'",
+ "span_status": "'not_started'",
+ "clear_number": "0",
+ },
+ "task_instance": {
+ "id": f"'{SEED_TASK_INSTANCE_ID}'",
+ "task_id": f"'{SEED_TASK_ID}'",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "map_index": "-1",
+ "pool": "'default'",
+ "pool_slots": "1",
+ "state": "'success'",
+ "try_number": "1",
+ "max_tries": "0",
+ },
+}
+
+# FK-dependency-respecting insertion order.
+SEED_ORDER = ("log_template", "dag_bundle", "dag", "dag_version", "dag_run", "task_instance")
+
+
+def _make_fresh_db() -> Path:
+ """Create a fresh SQLite file and re-point Airflow's engine at it."""
+ fd, path = tempfile.mkstemp(prefix="round_trip_", suffix=".db")
+ os.close(fd)
+ db = Path(path)
+ db.unlink() # Airflow creates the file on first connect
+
+ new_conn = f"sqlite:///{db}"
+ os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = new_conn
+
+ from airflow import settings
+
+ settings.SQL_ALCHEMY_CONN = new_conn
+ settings.dispose_orm()
+ settings.configure_orm()
+ return db
+
+
+def _step(msg: str) -> None:
+ console.print()
+ console.print(Rule(f"[bold cyan]{msg}[/]", style="cyan"))
+
+
+def _table_info(conn, table: str) -> list[tuple[Any, ...]]:
+ return list(conn.execute(text(f"PRAGMA table_info('{table}')")).fetchall())
+
+
+def _table_exists(conn, table: str) -> bool:
+ return bool(
+ conn.execute(
+ text("SELECT 1 FROM sqlite_master WHERE type='table' AND name=:n"),
+ {"n": table},
+ ).scalar()
+ )
+
+
+def _insert_or_skip(conn, table: str) -> None:
+ """Insert a seed row into ``table`` if the table exists and the row is missing.
+
+ Only columns present in the live schema and listed in ``SEED_VALUES`` are
+ included. Raises if a NOT-NULL-no-default column at the live revision is
+ not in ``SEED_VALUES``.
+ """
+ if not _table_exists(conn, table):
+ return
+
+ info = _table_info(conn, table)
+ live_cols = {r[1] for r in info}
+ not_null_no_default = {r[1] for r in info if r[3] and r[4] is None}
+
+ seed = SEED_VALUES[table]
+ use = {k: v for k, v in seed.items() if k in live_cols}
+ missing = not_null_no_default - set(use)
+ if missing:
+ raise RuntimeError(
+ f"Cannot seed `{table}`: NOT-NULL columns without defaults are missing seed values: "
+ f"{sorted(missing)}. Add them to SEED_VALUES['{table}'] in this script."
+ )
+
+ key = "dag_id" if "dag_id" in use else ("id" if "id" in use else next(iter(use)))
+ if conn.execute(text(f"SELECT 1 FROM {table} WHERE {key}={use[key]}")).scalar():
+ return
+
+ cols = ", ".join(use)
+ vals = ", ".join(use.values())
+ conn.execute(text(f"INSERT INTO {table} ({cols}) VALUES ({vals})"))
+
+
+def _seed_for_current_schema() -> None:
+ """Idempotently seed every table the live schema can accept.
+
+ Run before each migration step so the seed is in place to fire FK-cascade
+ chains during the rebuild that step performs (if any).
+ """
Review Comment:
`_seed_for_current_schema()` is called before every migration step; importing `airflow.settings` inside it is unnecessary and makes the “delayed import” intent unclear. Consider importing `settings` once (after the DB conn is set up) and passing the engine/settings object into seeding helpers, or at least document why this must be imported here.
```suggestion
chains during the rebuild that step performs (if any).
The ``airflow.settings`` import is intentionally delayed: this helper runs
after the migration step has configured Airflow's database state for the
current revision, and it must use the engine associated with that current
configuration rather than a module imported earlier during process startup.
"""
# Intentionally import here so seeding always uses the Airflow engine
# configured for the current migration step.
```
##########
scripts/in_container/run_migration_round_trip.py:
##########
@@ -0,0 +1,336 @@
+#!/usr/bin/env python
+# 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.
+"""
+Migration round-trip regression check — intended to run inside Breeze.
+
+Walks every revision base → head (upgrade direction) and head → base
+(downgrade direction) one rev at a time, on a single fresh SQLite DB
+with ``PRAGMA foreign_keys = ON``. Before each step the seed fixture is
+idempotently restored so any FK chain a rebuild needs to fire is in
+place.
+
+Per-migration walking is essential. ``airflow_db.upgradedb`` /
+``airflow_db.downgrade`` reconfigures Airflow's engine via
+``_single_connection_pool`` for every call, so each migration starts
+on a freshly-connected SQLAlchemy engine where ``setup_event_handlers``
+fires ``PRAGMA foreign_keys=ON``. In contrast a single-call
+``upgradedb(to_revision=tip)`` walks all migrations on one shared
+connection: once an earlier migration's ``disable_sqlite_fkeys`` flips
+FK off on that connection, the off state simply persists into later
+migrations, silently masking broken or missing wrappers downstream.
+
+The placement convention every migration must follow (only DML breaks
+PRAGMA, but the safe rule is strictly stronger) is documented in
+``contributing-docs/26_migration_round_trip_check.rst``.
+
+For the placement convention every migration must follow and the FK
+chains exercised by the seed, see
+``contributing-docs/26_migration_round_trip_check.rst``.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from alembic.script import ScriptDirectory
+from rich.console import Console
+from rich.panel import Panel
+from rich.rule import Rule
+from sqlalchemy import text
+
+console = Console(width=140, color_system="standard", highlight=False)
+
+SEED_DAG_ID = "round_trip_dag"
+SEED_DAG_VERSION_ID = "abcdef0123456789abcdef0123456789"
+SEED_BUNDLE_NAME = "dags-folder"
+SEED_RUN_ID = "manual_run_1"
+SEED_TASK_ID = "task1"
+SEED_TASK_INSTANCE_ID = "fedcba9876543210fedcba9876543210"
+
+# Hardcoded INSERT values for tables we seed. Each entry maps column name
+# to a SQL literal. When a column is not present in the live schema (e.g.
+# an older revision predating its addition), it is dropped from the INSERT
+# — provided every NOT-NULL-no-default column at that revision is covered.
+# When a future migration introduces a brand-new NOT NULL column without a
+# default, ``_insert_or_skip`` raises with a clear message pointing at the
+# table that needs an entry here.
+SEED_VALUES: dict[str, dict[str, str]] = {
+ "log_template": {
+ "id": "1",
+ "filename": "'{{ ti.dag_id }}/{{ ts }}.log'",
+ "elasticsearch_id": "'{{ ti.dag_id }}-{{ ts }}'",
+ "created_at": "'2024-01-01 00:00:00.000000+00:00'",
+ },
+ "dag_bundle": {
+ "name": f"'{SEED_BUNDLE_NAME}'",
+ "active": "1",
+ "version": "NULL",
+ "last_refreshed": "NULL",
+ "signed_url_template": "NULL",
+ "url_template": "NULL",
+ "bundle_template_params": "NULL",
+ },
+ "dag": {
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "max_active_tasks": "16",
+ "has_task_concurrency_limits": "0",
+ "is_paused": "0",
+ "is_active": "1",
+ "is_stale": "0",
+ "fileloc": "'/tmp/round_trip_dag.py'",
+ "fail_fast": "0",
+ "max_consecutive_failed_dag_runs": "0",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "has_import_errors": "0",
+ "max_active_runs": "16",
+ "_default_view": "'grid'",
+ "exceeds_max_non_backfill": "0",
+ "timetable_type": "'cron'",
+ "timetable_partitioned": "0",
+ "timetable_periodic": "0",
+ },
+ "dag_version": {
+ "id": f"'{SEED_DAG_VERSION_ID}'",
+ "version_number": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "bundle_name": f"'{SEED_BUNDLE_NAME}'",
+ "bundle_version": "NULL",
+ "created_at": "'2024-01-01 00:00:00+00:00'",
+ "last_updated": "'2024-01-01 00:00:00+00:00'",
+ },
+ "dag_run": {
+ "id": "1",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "execution_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "logical_date": "'2024-01-01 00:00:00.000000+00:00'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "run_type": "'manual'",
+ "state": "'success'",
+ "log_template_id": "1",
+ "run_after": "'2024-01-01 00:00:00+00:00'",
+ "span_status": "'not_started'",
+ "clear_number": "0",
+ },
+ "task_instance": {
+ "id": f"'{SEED_TASK_INSTANCE_ID}'",
+ "task_id": f"'{SEED_TASK_ID}'",
+ "dag_id": f"'{SEED_DAG_ID}'",
+ "run_id": f"'{SEED_RUN_ID}'",
+ "map_index": "-1",
+ "pool": "'default'",
+ "pool_slots": "1",
+ "state": "'success'",
+ "try_number": "1",
+ "max_tries": "0",
+ },
+}
+
+# FK-dependency-respecting insertion order.
+SEED_ORDER = ("log_template", "dag_bundle", "dag", "dag_version", "dag_run", "task_instance")
+
+
+def _make_fresh_db() -> Path:
+ """Create a fresh SQLite file and re-point Airflow's engine at it."""
+ fd, path = tempfile.mkstemp(prefix="round_trip_", suffix=".db")
+ os.close(fd)
+ db = Path(path)
+ db.unlink() # Airflow creates the file on first connect
+
+ new_conn = f"sqlite:///{db}"
+ os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = new_conn
+
+ from airflow import settings
+
+ settings.SQL_ALCHEMY_CONN = new_conn
+ settings.dispose_orm()
+ settings.configure_orm()
+ return db
Review Comment:
Importing `airflow.settings` inside this helper violates the repo’s general “imports at top of file” rule. If this import is intentionally delayed so `AIRFLOW__DATABASE__SQL_ALCHEMY_CONN` can be set before Airflow config is loaded, please add an explicit comment explaining that constraint (or refactor so the import happens once in a clearly-delayed, top-level initialization path).
--
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] ci: add migration round-trip check for SQLite FK cascade regressions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #65547:
URL: https://github.com/apache/airflow/pull/65547#issuecomment-4371253277
> Another layer of defense?
yep
> Should we update those rules (or add another one) to capture the issue in 108?
kinda feel we can keep it as it is. and do it if we encounter it next time
--
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