You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "arjav1528 (via GitHub)" <gi...@apache.org> on 2026/01/11 13:27:49 UTC
[PR] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
arjav1528 opened a new pull request, #60390:
URL: https://github.com/apache/airflow/pull/60390
## Problem
When using `airflow db clean`, an error occurs when deleting rows from the `dag_version` table due to a violation of the `task_instance_dag_version_id_fkey` foreign key constraint.
The issue occurs when:
- A `dag_version` row has an old `created_at` timestamp (meets deletion criteria)
- A `task_instance` row has a recent `start_date` timestamp (does NOT meet deletion criteria)
- The `task_instance` row references the old `dag_version` row
## Fixes: #59474
The `db clean` command was removing rows from both tables based solely on their respective recency columns (`created_at` for `dag_version`, `start_date` for `task_instance`) without considering the foreign key relationship.
## Solution
Modified `_build_query()` in `db_cleanup.py` to add special handling for the `dag_version` table. When building the deletion query for `dag_version`, we now:
1. Create a subquery to find `dag_version_id`s that are referenced by `task_instance` rows with `start_date >= clean_before_timestamp` (i.e., rows that are NOT being deleted)
2. Exclude those `dag_version` rows from the deletion query
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r2694257187
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
+ except Exception:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ pass
Review Comment:
> Is there any reason that we can fail here and it is not a problem? What cause can it have to fail? I would rather prefer to fail "in general" or "fail fast" to fix bugs and not swallowing the exception.
On second thought, I think we should only be swallowing classes of exceptions (with a logged warning of course) that we can expect to arise during normal operation and that are not indicative of more critical issues like an unhealthy DB or logic errors in the code (broad exceptions can mask regressions). The latter class of errors do need to fail loudly and stop the operation.
I believe the auto-reviewer’s suggestion below (Except SQLAlchemyError and KeyError) is in the right direction. But I think SQLAlchemyError should be further narrowed to exclude OperationalError and the like.
I have no issues with the ‘fail fast’ and ‘in general’ approach you are suggesting. But your comment made me re-evaluate whether catching all exceptions is acceptable.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-4035152066
@arjav1528 This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Gen-AI disclosure**: GitHub Copilot is listed as a co-author on three commits in this PR (commits updating db_cleanup.py and check_secrets_search_path_sync.py both carry 'Co-authored-by: Copilot'), yet the PR description contains no Gen-AI disclosure checkbox or statement. The contributing guidelines explicitly require: 'State in your PR description that you have used Gen-AI tools to assist in creating the PR.'
- :x: **Coherent changes**: The file `scripts/ci/prek/check_secrets_search_path_sync.py` is modified (adding `and isinstance(elt.value, str)` to an AST traversal check) but is entirely unrelated to the stated purpose of the PR (fixing an FK constraint violation in `airflow db clean` for the `dag_version` table). This is a classic AI-generated-code signal: a tool introduced an unrelated change that the contributor did not remove before submitting. The contributing guidelines explicitly warn: 'Review your code to make sure it does not contain unrelated changes. Often Gen-AI tools might introduce changes that are not related to the problem you are trying to solve.'
> **Note:** Your branch is **1214 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.
**What to do next:**
- The comment informs you what you need to do.
- Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
- Maintainers will then proceed with a normal review.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-4387607019
@arjav1528 This draft PR has been inactive for 13 days since the last triage comment and no response from the author. Closing to keep the queue clean.
You are welcome to reopen this PR when you resume work, or to open a new one addressing the issues previously raised. There is no rush — take your time.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "arjav1528 (via GitHub)" <gi...@apache.org>.
arjav1528 commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3794760623
> eyes
are there any changes left??
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "m1racoli (via GitHub)" <gi...@apache.org>.
m1racoli commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3923947138
FYI There is https://github.com/apache/airflow/pull/62023 open to fix apparently the same problem. But instead introducing custom handling of the `dag_version` table, it introduces the concept of "skip if referenced" which may be useful for further tables in the future. I am wondering if it is worth considering.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r3066477523
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -741,6 +741,318 @@ def test_drop_archived_tables(self, mock_input, confirm_mock, inspect_mock, capl
else:
confirm_mock.assert_not_called()
+ def test_dag_version_cleanup_respects_task_instance_fk_constraint(self):
+ """
+ Verify that dag_version cleanup does not delete dag_version rows that are
+ referenced by task_instance rows that are being kept (not deleted).
+
+ This tests the FK constraint handling added to prevent:
+ - task_instance.dag_version_id uses ondelete="RESTRICT", so deleting a
+ dag_version that's still referenced by a kept task_instance would fail.
+
+ The test creates:
+ - 2 dag_versions (old and new)
+ - 2 task_instances: one old (to be deleted) and one new (to be kept)
+ - Both TIs reference different dag_versions
+
+ When cleaning up with a timestamp that keeps the new TI, the dag_version
+ referenced by the kept TI should NOT be deleted, even if it's old.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-constraint"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+
+ # Create an old dag_run and task_instance (to be deleted)
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create a new dag_run and task_instance (to be kept)
+ new_start_date = base_date.add(days=10)
+ new_dag_run = DagRun(
+ dag.dag_id,
+ run_id="new_run",
+ run_type=DagRunType.MANUAL,
+ start_date=new_start_date,
+ )
+ new_ti = create_task_instance(
+ PythonOperator(task_id="new-task", python_callable=print),
+ run_id=new_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ new_ti.dag_id = dag.dag_id
+ new_ti.start_date = new_start_date
+ session.add(new_dag_run)
+ session.add(new_ti)
+ session.commit()
+
+ # Clean before timestamp that would keep the new TI but delete the old one
+ clean_before_date = base_date.add(days=5)
+
+ # Count dag_versions before cleanup
+ dag_versions_before = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_versions_before == 1
+
+ # Run cleanup on dag_version table
+ # The dag_version is old (created_at < clean_before_date), but it's referenced
+ # by a kept task_instance (new_ti with start_date >= clean_before_date)
+ # So it should NOT be deleted due to the FK constraint handling
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
Review Comment:
This test does not currently exercise the reported failure mode. `dag_version` cleanup uses `created_at` plus `keep_last=True`; with only a single `dag_version` created here (and without forcing its `created_at` to be older than `clean_before_date` and also not the latest), the row won’t be eligible for deletion even without the new FK-aware filter. To validate the fix, create at least two `dag_version` rows for the same `dag_id`, make the *older* one eligible for deletion (`created_at < clean_before_date` and not the latest), and have a kept `task_instance` reference that older version; then assert the older `dag_version` is excluded from deletion.
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -741,6 +741,318 @@ def test_drop_archived_tables(self, mock_input, confirm_mock, inspect_mock, capl
else:
confirm_mock.assert_not_called()
+ def test_dag_version_cleanup_respects_task_instance_fk_constraint(self):
+ """
+ Verify that dag_version cleanup does not delete dag_version rows that are
+ referenced by task_instance rows that are being kept (not deleted).
+
+ This tests the FK constraint handling added to prevent:
+ - task_instance.dag_version_id uses ondelete="RESTRICT", so deleting a
+ dag_version that's still referenced by a kept task_instance would fail.
+
+ The test creates:
+ - 2 dag_versions (old and new)
+ - 2 task_instances: one old (to be deleted) and one new (to be kept)
+ - Both TIs reference different dag_versions
+
+ When cleaning up with a timestamp that keeps the new TI, the dag_version
+ referenced by the kept TI should NOT be deleted, even if it's old.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-constraint"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+
+ # Create an old dag_run and task_instance (to be deleted)
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create a new dag_run and task_instance (to be kept)
+ new_start_date = base_date.add(days=10)
+ new_dag_run = DagRun(
+ dag.dag_id,
+ run_id="new_run",
+ run_type=DagRunType.MANUAL,
+ start_date=new_start_date,
+ )
+ new_ti = create_task_instance(
+ PythonOperator(task_id="new-task", python_callable=print),
+ run_id=new_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ new_ti.dag_id = dag.dag_id
+ new_ti.start_date = new_start_date
+ session.add(new_dag_run)
+ session.add(new_ti)
+ session.commit()
+
+ # Clean before timestamp that would keep the new TI but delete the old one
+ clean_before_date = base_date.add(days=5)
+
+ # Count dag_versions before cleanup
+ dag_versions_before = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_versions_before == 1
+
+ # Run cleanup on dag_version table
+ # The dag_version is old (created_at < clean_before_date), but it's referenced
+ # by a kept task_instance (new_ti with start_date >= clean_before_date)
+ # So it should NOT be deleted due to the FK constraint handling
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The query should return 0 rows to delete because the dag_version
+ # is referenced by a kept task_instance
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 0, (
+ f"Expected 0 dag_version rows to be marked for deletion, got {rows_to_delete}. "
+ "The dag_version should be protected because it's referenced by a kept task_instance."
+ )
+
+ def test_dag_version_cleanup_deletes_unreferenced_rows(self):
+ """
+ Verify that dag_version rows that are NOT referenced by any task_instance
+ are properly deleted during cleanup.
+
+ This ensures the FK constraint handling doesn't over-protect dag_version rows
+ that have no task_instance references at all.
+
+ Note: dag_version has keep_last=True, so the most recent dag_version per dag_id
+ is always kept. We need at least 2 dag_versions for a dag to test deletion.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ # Create a dag with TWO dag_versions (neither has TI references)
+ # The older one should be deleted, the newer one kept by keep_last
+ dag_id = "test-dag-no-ti"
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ session.flush()
+
+ # Create first (old) dag_version using DagVersion.write_dag directly
+ old_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ old_dag_version.created_at = base_date # Old date
+ session.flush()
+
+ # Create second (new) dag_version
+ new_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ new_dag_version.created_at = base_date.add(days=10) # New date
+ session.commit()
+
+ # Verify we have 2 dag_versions
+ dag_version_count = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_version_count == 2, f"Expected 2 dag_versions, got {dag_version_count}"
+
+ # Clean before timestamp that would delete the old dag_version
+ clean_before_date = base_date.add(days=5)
+
+ # Build query for dag_version cleanup
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The old unreferenced dag_version should be marked for deletion
+ # (the new one is kept by keep_last)
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 1, (
+ f"Expected 1 dag_version row to be marked for deletion, got {rows_to_delete}. "
+ "The old unreferenced dag_version should be deleted."
+ )
+
+ def test_dag_version_cleanup_deletes_rows_only_referenced_by_old_ti(self):
+ """
+ Verify that dag_version rows that are only referenced by OLD task_instance
+ rows (that will be deleted) can be properly deleted.
+
+ After task_instance cleanup removes old TIs, the dag_version rows they
+ referenced should become eligible for deletion.
+
+ Note: dag_version has keep_last=True, so we need 2 dag_versions. The older
+ one (referenced only by old TI) should be deleted, the newer one is kept.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-old-ti-only"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ session.flush()
+
+ # Create first (old) dag_version using DagVersion.write_dag directly
+ old_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ old_dag_version.created_at = base_date # Old date
+ session.flush()
+
+ # Create ONLY an old task_instance referencing the OLD dag_version
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=old_dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create second (new) dag_version - no TI references this one
+ new_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ new_dag_version.created_at = base_date.add(days=10) # New date, kept by keep_last
+ session.commit()
+
+ # Verify we have 2 dag_versions
+ dag_version_count = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_version_count == 2, f"Expected 2 dag_versions, got {dag_version_count}"
+
+ # Clean before timestamp that would delete the old TI
+ clean_before_date = base_date.add(days=5)
+
+ # Build query for dag_version cleanup
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The OLD dag_version should be marked for deletion since:
+ # 1. Its created_at is old (base_date < clean_before_date)
+ # 2. The only task_instance referencing it is old (will also be deleted)
+ # 3. It's not the latest dag_version (keep_last keeps the new one)
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 1, (
+ f"Expected 1 dag_version row to be marked for deletion, got {rows_to_delete}. "
+ "The dag_version referenced only by old TIs should be deleted."
+ )
+
+ def test_dag_version_cleanup_no_fk_violation_integration(self):
+ """
+ Integration test to verify that the full cleanup process doesn't fail
+ with FK violations when task_instance and dag_version have different
+ recency timestamps.
+
+ This tests the actual run_cleanup flow, not just the query building.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-integration"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+ # Set dag_version created_at to old date (eligible for cleanup by timestamp)
+ dag_version.created_at = base_date
+
+ # Create a NEW task_instance that references this OLD dag_version
+ # This is the scenario that would cause FK violation without proper handling
+ new_start_date = base_date.add(days=10)
+ dag_run = DagRun(
+ dag.dag_id,
+ run_id="new_run",
+ run_type=DagRunType.MANUAL,
+ start_date=new_start_date,
+ )
+ ti = create_task_instance(
+ PythonOperator(task_id="new-task", python_callable=print),
+ run_id=dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ ti.dag_id = dag.dag_id
+ ti.start_date = new_start_date
+ session.add(dag_run)
+ session.add(ti)
+ session.commit()
+
+ # Clean before timestamp that keeps the new TI but would normally
+ # try to delete the old dag_version (based on created_at)
+ clean_before_date = base_date.add(days=5)
+
+ # Count dag_versions before cleanup
+ dag_versions_before = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+
+ # Run the full cleanup - this should NOT raise FK violation errors
+ # The dag_version table cleanup should skip this row because it's
+ # referenced by a kept task_instance
+ run_cleanup(
+ clean_before_timestamp=clean_before_date,
+ table_names=["dag_version"],
+ dry_run=False,
+ confirm=False,
+ session=session,
+ )
Review Comment:
This integration test is unlikely to catch the FK constraint regression because it only creates a single `dag_version` for the DAG. With `keep_last=True`, the only version is always kept, so `run_cleanup(table_names=[\"dag_version\"])` won’t attempt to delete it regardless of the FK-aware filtering. To meaningfully validate the fix end-to-end, create two `dag_version` rows and ensure the older one would be deleted by timestamp/keep_last logic, while a kept `task_instance` still references that older version; then assert the cleanup completes and the referenced older version remains.
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -372,6 +373,49 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted.
+ #
+ # dag_version has two dependent tables (see config_list):
+ # - task_instance.dag_version_id: uses ondelete="RESTRICT" - will BLOCK deletion if referenced
+ # - dag_run.created_dag_version_id: uses ondelete="SET NULL" - automatically sets to NULL on delete
+ #
+ # We only need to handle task_instance here because dag_run's FK constraint won't cause violations
+ # (the database handles it automatically by setting the FK to NULL when dag_version is deleted).
+ if table_name == "dag_version":
+ try:
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ # Use EXISTS for better performance and NULL handling
+ base_table_id_col = base_table.c[dv_id_col.name]
+ kept_tis_exists = exists(
+ select(1)
+ .select_from(ti_table_reflected)
+ .where(ti_dag_version_id_col == base_table_id_col)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Negate EXISTS to get NOT EXISTS behavior
+ conditions.append(~kept_tis_exists)
+ except (KeyError, AttributeError, OperationalError, ProgrammingError) as e:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ logger.warning(
+ "Failed to add foreign key constraint filter for dag_version table cleanup: %s. "
+ "Continuing without the filter. Note that FK violations may still be encountered during cleanup.",
+ type(e).__name__,
+ )
Review Comment:
Continuing without the FK filter can reintroduce the exact FK violation this PR is addressing, causing `db clean` to fail later during delete execution. A safer fallback would be to skip `dag_version` deletions in this run (e.g., add a condition that yields no rows) when the filter can’t be constructed. Also, the warning logs only the exception type; including the exception message and/or `exc_info=True` would make failures diagnosable.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #60390: Fix FK constraint violation in `airflow db clean` for dag_version table
URL: https://github.com/apache/airflow/pull/60390
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r2692143845
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
Review Comment:
According to the dag_version configuration in config_list (line 167), dag_version has dependent_tables including both "task_instance" and "dag_run". While dag_run.created_dag_version_id has ondelete="set null" (which won't cause FK violations), the current implementation only handles the task_instance constraint. Consider whether dag_run also needs to be checked, or add a comment explaining why only task_instance requires special handling (i.e., because dag_run uses ondelete="set null").
##########
scripts/ci/prek/check_secrets_search_path_sync.py:
##########
@@ -42,9 +42,9 @@ def extract_from_file(file_path: Path, constant_name: str) -> list[str] | None:
if isinstance(node.value, ast.List):
values = []
for elt in node.value.elts:
- if isinstance(elt, ast.Constant):
+ if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
values.append(elt.value)
- return values
+ return values if values else None
Review Comment:
This change appears unrelated to the PR's stated purpose of fixing FK constraint violations in airflow db clean for the dag_version table. While the changes are valid improvements (filtering non-string constants and returning None for empty lists), they should ideally be in a separate PR focused on improving the check_secrets_search_path_sync script.
```suggestion
return values
```
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
+ except Exception:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ pass
+
Review Comment:
The new dag_version foreign key constraint handling logic lacks test coverage. Consider adding test cases that verify: 1) dag_version rows referenced by recent task_instance rows are NOT deleted, 2) dag_version rows not referenced by any task_instance (or only by old task_instance rows) ARE deleted, and 3) the cleanup doesn't fail with FK violations when task_instance and dag_version have different recency timestamps.
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
Review Comment:
The import statement for reflect_tables is redundant since it's already imported at the top of the file (line 43). The local import inside the try block should be removed.
```suggestion
```
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
+ except Exception:
Review Comment:
The bare except clause catches all exceptions including system exceptions like KeyboardInterrupt and SystemExit. This should be narrowed to catch only the relevant exceptions (e.g., SQLAlchemyError, KeyError) that might occur during table reflection or column access. The current implementation could silently swallow critical errors.
```suggestion
except (KeyError, AttributeError, OperationalError, ProgrammingError):
```
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "arjav1528 (via GitHub)" <gi...@apache.org>.
arjav1528 commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3754704427
@SameerMesiah97 I have applied the changes, do review them at your availability\
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r2692109873
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
+ except Exception:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ pass
Review Comment:
Is there any reason that we can fail here and it is not a problem? What cause can it have to fail?
I would rather prefer to fail "in general" or "fail fast" to fix bugs and not swallowing the exception.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "arjav1528 (via GitHub)" <gi...@apache.org>.
arjav1528 commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3762980956
> Thanks, looks good now in my view. Prior merge, would like to see another pair of (critical) eyes on the code.
Sure, would be happy to recieve suggestions
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-4300877561
@arjav1528 This draft PR has not had author activity for 21 days. Before anyone can review it, it needs to meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria). Here is a summary of what still needs work; once those items are addressed, mark the PR ready for review.
**Issues found:**
- :x: **Failing CI**: `Additional PROD image tests / Firefox UI e2e tests with PROD image / Firefox UI e2e tests`, `Additional PROD image tests / Test e2e integration tests with PROD image / Regular e2e test`. Please investigate and fix.
- :x: **Unresolved review comments** (9 threads): please walk through each unresolved review thread. **Even if a suggestion looks incorrect or irrelevant** — and some of them will be, especially any comments left by automated reviewers like GitHub Copilot — it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. **Once you believe a thread is resolved — whether by pushing a fix or by explaining why the suggestion doesn't apply — please mark it as resolved yourself by clicking the 'Resolve conversation' button at the bottom of each thread.** Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as 'still waiting on the author' and keeps the PR from moving forward.
**What to do next:**
- Address the items above.
- Make sure static checks pass locally: `prek run --from-ref main --stage pre-commit`.
- Mark the PR as "Ready for review" when you are done.
There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "arjav1528 (via GitHub)" <gi...@apache.org>.
arjav1528 commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3747945905
@maintainers, could you please review the PR
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r2684029069
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
Review Comment:
I understand you used DISTINCT because multiple task instances can reference the same `dag_version_id` and you wanted to de-duplicate them but I believe it would be better to use a NOT EXISTS approach here for better performance and NULL handling. kept_ti_subquery (you can rename it to kept_tis_exists for more clarity) could be instantiated with the EXISTS ORM query which you can then negate using '~' when you append it to the conditions (which would be evaluated as NOT EXISTS). This would have the same effect as de-depulicating using DISTINCT and using NOT IN.
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -365,6 +366,39 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted
+ if table_name == "dag_version":
+ try:
+ from airflow.utils.db import reflect_tables
+
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ kept_ti_subquery = (
+ select(ti_dag_version_id_col)
+ .select_from(ti_table_reflected)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ .distinct()
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Use the reflected table's id column and join it with base_table
+ base_table_id_col = base_table.c[dv_id_col.name]
+ conditions.append(base_table_id_col.not_in(kept_ti_subquery))
+ except Exception:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ pass
Review Comment:
The Exception is being swallowed silently. I would suggest naming it in a `logger.warning` message (please see other examples in this module) alongside text that clearly informs the user that FK violations may still be encountered.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #60390:
URL: https://github.com/apache/airflow/pull/60390#issuecomment-3750862758
> @maintainers, could you please review the PR
I have left a few comments that you could address whilst waiting for an authoritative review.
--
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] Fix FK constraint violation in `airflow db clean` for dag_version table [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60390:
URL: https://github.com/apache/airflow/pull/60390#discussion_r2733030893
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -371,6 +372,49 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted.
+ #
+ # dag_version has two dependent tables (see config_list):
+ # - task_instance.dag_version_id: uses ondelete="RESTRICT" - will BLOCK deletion if referenced
+ # - dag_run.created_dag_version_id: uses ondelete="SET NULL" - automatically sets to NULL on delete
+ #
+ # We only need to handle task_instance here because dag_run's FK constraint won't cause violations
+ # (the database handles it automatically by setting the FK to NULL when dag_version is deleted).
+ if table_name == "dag_version":
+ try:
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ # Use EXISTS for better performance and NULL handling
+ base_table_id_col = base_table.c[dv_id_col.name]
+ kept_tis_exists = exists(
+ select(1)
+ .select_from(ti_table_reflected)
+ .where(ti_dag_version_id_col == base_table_id_col)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ )
Review Comment:
The FK-protection EXISTS only treats task_instance rows with `start_date >= clean_before_timestamp` as “kept”. But `task_instance.start_date` is nullable, and rows with `start_date IS NULL` are also kept by the cleanup query (`NULL < timestamp` is not true). Those NULL-start_date task_instances can still reference dag_version and will trigger FK violations unless they’re included in the EXISTS predicate (e.g., treat `start_date IS NULL` as kept, or mirror the exact task_instance deletion criteria).
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -741,6 +741,318 @@ def test_drop_archived_tables(self, mock_input, confirm_mock, inspect_mock, capl
else:
confirm_mock.assert_not_called()
+ def test_dag_version_cleanup_respects_task_instance_fk_constraint(self):
+ """
+ Verify that dag_version cleanup does not delete dag_version rows that are
+ referenced by task_instance rows that are being kept (not deleted).
+
+ This tests the FK constraint handling added to prevent:
+ - task_instance.dag_version_id uses ondelete="RESTRICT", so deleting a
+ dag_version that's still referenced by a kept task_instance would fail.
+
+ The test creates:
+ - 2 dag_versions (old and new)
+ - 2 task_instances: one old (to be deleted) and one new (to be kept)
+ - Both TIs reference different dag_versions
+
+ When cleaning up with a timestamp that keeps the new TI, the dag_version
+ referenced by the kept TI should NOT be deleted, even if it's old.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-constraint"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+
Review Comment:
This test doesn’t currently exercise the FK-protection behavior: it only creates a single dag_version (`DagVersion.get_latest_version(...)`), and `dag_version` cleanup is configured with `keep_last=True`, so the lone (latest) version won’t be selected for deletion regardless of whether any task_instance references it. To validate the fix, create at least two dag_versions for the dag, make the older one eligible by `created_at < clean_before_date` and not the latest, and have a kept task_instance reference that older version.
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -741,6 +741,318 @@ def test_drop_archived_tables(self, mock_input, confirm_mock, inspect_mock, capl
else:
confirm_mock.assert_not_called()
+ def test_dag_version_cleanup_respects_task_instance_fk_constraint(self):
+ """
+ Verify that dag_version cleanup does not delete dag_version rows that are
+ referenced by task_instance rows that are being kept (not deleted).
+
+ This tests the FK constraint handling added to prevent:
+ - task_instance.dag_version_id uses ondelete="RESTRICT", so deleting a
+ dag_version that's still referenced by a kept task_instance would fail.
+
+ The test creates:
+ - 2 dag_versions (old and new)
+ - 2 task_instances: one old (to be deleted) and one new (to be kept)
+ - Both TIs reference different dag_versions
+
+ When cleaning up with a timestamp that keeps the new TI, the dag_version
+ referenced by the kept TI should NOT be deleted, even if it's old.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-constraint"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+
+ # Create an old dag_run and task_instance (to be deleted)
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create a new dag_run and task_instance (to be kept)
+ new_start_date = base_date.add(days=10)
+ new_dag_run = DagRun(
+ dag.dag_id,
+ run_id="new_run",
+ run_type=DagRunType.MANUAL,
+ start_date=new_start_date,
+ )
+ new_ti = create_task_instance(
+ PythonOperator(task_id="new-task", python_callable=print),
+ run_id=new_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ new_ti.dag_id = dag.dag_id
+ new_ti.start_date = new_start_date
+ session.add(new_dag_run)
+ session.add(new_ti)
+ session.commit()
+
+ # Clean before timestamp that would keep the new TI but delete the old one
+ clean_before_date = base_date.add(days=5)
+
+ # Count dag_versions before cleanup
+ dag_versions_before = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_versions_before == 1
+
+ # Run cleanup on dag_version table
+ # The dag_version is old (created_at < clean_before_date), but it's referenced
+ # by a kept task_instance (new_ti with start_date >= clean_before_date)
+ # So it should NOT be deleted due to the FK constraint handling
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The query should return 0 rows to delete because the dag_version
+ # is referenced by a kept task_instance
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 0, (
+ f"Expected 0 dag_version rows to be marked for deletion, got {rows_to_delete}. "
+ "The dag_version should be protected because it's referenced by a kept task_instance."
+ )
+
+ def test_dag_version_cleanup_deletes_unreferenced_rows(self):
+ """
+ Verify that dag_version rows that are NOT referenced by any task_instance
+ are properly deleted during cleanup.
+
+ This ensures the FK constraint handling doesn't over-protect dag_version rows
+ that have no task_instance references at all.
+
+ Note: dag_version has keep_last=True, so the most recent dag_version per dag_id
+ is always kept. We need at least 2 dag_versions for a dag to test deletion.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ # Create a dag with TWO dag_versions (neither has TI references)
+ # The older one should be deleted, the newer one kept by keep_last
+ dag_id = "test-dag-no-ti"
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ session.flush()
+
+ # Create first (old) dag_version using DagVersion.write_dag directly
+ old_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ old_dag_version.created_at = base_date # Old date
+ session.flush()
+
+ # Create second (new) dag_version
+ new_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ new_dag_version.created_at = base_date.add(days=10) # New date
+ session.commit()
+
+ # Verify we have 2 dag_versions
+ dag_version_count = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_version_count == 2, f"Expected 2 dag_versions, got {dag_version_count}"
+
+ # Clean before timestamp that would delete the old dag_version
+ clean_before_date = base_date.add(days=5)
+
+ # Build query for dag_version cleanup
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The old unreferenced dag_version should be marked for deletion
+ # (the new one is kept by keep_last)
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 1, (
+ f"Expected 1 dag_version row to be marked for deletion, got {rows_to_delete}. "
+ "The old unreferenced dag_version should be deleted."
+ )
+
+ def test_dag_version_cleanup_deletes_rows_only_referenced_by_old_ti(self):
+ """
+ Verify that dag_version rows that are only referenced by OLD task_instance
+ rows (that will be deleted) can be properly deleted.
+
+ After task_instance cleanup removes old TIs, the dag_version rows they
+ referenced should become eligible for deletion.
+
+ Note: dag_version has keep_last=True, so we need 2 dag_versions. The older
+ one (referenced only by old TI) should be deleted, the newer one is kept.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-old-ti-only"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ session.flush()
+
+ # Create first (old) dag_version using DagVersion.write_dag directly
+ old_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ old_dag_version.created_at = base_date # Old date
+ session.flush()
+
+ # Create ONLY an old task_instance referencing the OLD dag_version
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=old_dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create second (new) dag_version - no TI references this one
+ new_dag_version = DagVersion.write_dag(dag_id=dag_id, bundle_name=bundle_name, session=session)
+ new_dag_version.created_at = base_date.add(days=10) # New date, kept by keep_last
+ session.commit()
+
+ # Verify we have 2 dag_versions
+ dag_version_count = session.scalar(
+ select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag_id)
+ )
+ assert dag_version_count == 2, f"Expected 2 dag_versions, got {dag_version_count}"
+
+ # Clean before timestamp that would delete the old TI
+ clean_before_date = base_date.add(days=5)
+
+ # Build query for dag_version cleanup
+ query = _build_query(
+ **config_dict["dag_version"].__dict__,
+ clean_before_timestamp=clean_before_date,
+ session=session,
+ )
+
+ # The OLD dag_version should be marked for deletion since:
+ # 1. Its created_at is old (base_date < clean_before_date)
+ # 2. The only task_instance referencing it is old (will also be deleted)
+ # 3. It's not the latest dag_version (keep_last keeps the new one)
+ rows_to_delete = session.scalar(select(func.count()).select_from(query.subquery()))
+ assert rows_to_delete == 1, (
+ f"Expected 1 dag_version row to be marked for deletion, got {rows_to_delete}. "
+ "The dag_version referenced only by old TIs should be deleted."
+ )
+
+ def test_dag_version_cleanup_no_fk_violation_integration(self):
+ """
+ Integration test to verify that the full cleanup process doesn't fail
+ with FK violations when task_instance and dag_version have different
+ recency timestamps.
+
+ This tests the actual run_cleanup flow, not just the query building.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-integration"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+ # Set dag_version created_at to old date (eligible for cleanup by timestamp)
+ dag_version.created_at = base_date
+
Review Comment:
This integration test also only creates a single dag_version for the dag. Since dag_version cleanup uses `keep_last=True`, that latest version will be kept even when `created_at` is old, so the cleanup won’t attempt the delete that would trigger the FK violation. Consider creating two dag_versions and making the older one eligible for deletion + referenced by a kept task_instance, then assert the older one remains and no IntegrityError is raised.
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -371,6 +372,49 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted.
+ #
+ # dag_version has two dependent tables (see config_list):
+ # - task_instance.dag_version_id: uses ondelete="RESTRICT" - will BLOCK deletion if referenced
+ # - dag_run.created_dag_version_id: uses ondelete="SET NULL" - automatically sets to NULL on delete
+ #
+ # We only need to handle task_instance here because dag_run's FK constraint won't cause violations
+ # (the database handles it automatically by setting the FK to NULL when dag_version is deleted).
+ if table_name == "dag_version":
+ try:
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ # Use EXISTS for better performance and NULL handling
+ base_table_id_col = base_table.c[dv_id_col.name]
+ kept_tis_exists = exists(
+ select(1)
+ .select_from(ti_table_reflected)
+ .where(ti_dag_version_id_col == base_table_id_col)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Negate EXISTS to get NOT EXISTS behavior
+ conditions.append(~kept_tis_exists)
+ except (KeyError, AttributeError, OperationalError, ProgrammingError) as e:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
+ logger.warning(
+ "Failed to add foreign key constraint filter for dag_version table cleanup: %s. "
+ "Continuing without the filter. Note that FK violations may still be encountered during cleanup.",
+ type(e).__name__,
Review Comment:
The warning in the exception handler only logs the exception type, which makes diagnosing real FK-protection failures difficult (and in the current code path, `KeyError` is likely). Log the exception message and include traceback (`exc_info=True`) so operators can see why the filter could not be applied.
```suggestion
e,
exc_info=True,
```
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -371,6 +372,49 @@ def _build_query(
),
)
conditions.append(column(max_date_col_name).is_(None))
+
+ # Special handling for dag_version: exclude rows referenced by task_instance rows that are NOT being deleted.
+ #
+ # dag_version has two dependent tables (see config_list):
+ # - task_instance.dag_version_id: uses ondelete="RESTRICT" - will BLOCK deletion if referenced
+ # - dag_run.created_dag_version_id: uses ondelete="SET NULL" - automatically sets to NULL on delete
+ #
+ # We only need to handle task_instance here because dag_run's FK constraint won't cause violations
+ # (the database handles it automatically by setting the FK to NULL when dag_version is deleted).
+ if table_name == "dag_version":
+ try:
+ # Reflect the task_instance table to get proper column access
+ metadata = reflect_tables(["task_instance", "dag_version"], session)
+ ti_table_reflected = metadata.tables["task_instance"]
+ dv_table_reflected = metadata.tables["dag_version"]
+ ti_dag_version_id_col = ti_table_reflected.c.dag_version_id
+ ti_start_date_col = ti_table_reflected.c.start_date
+ dv_id_col = dv_table_reflected.c.id
+
+ # Find dag_version_ids that are referenced by task_instance rows that are kept (not deleted)
+ # These are task_instance rows with start_date >= clean_before_timestamp
+ # Use EXISTS for better performance and NULL handling
+ base_table_id_col = base_table.c[dv_id_col.name]
+ kept_tis_exists = exists(
+ select(1)
+ .select_from(ti_table_reflected)
+ .where(ti_dag_version_id_col == base_table_id_col)
+ .where(ti_start_date_col >= clean_before_timestamp)
+ .where(ti_dag_version_id_col.isnot(None))
+ )
+
+ # Exclude dag_version rows that are referenced by kept task_instance rows
+ # Negate EXISTS to get NOT EXISTS behavior
+ conditions.append(~kept_tis_exists)
+ except (KeyError, AttributeError, OperationalError, ProgrammingError) as e:
+ # If we can't add the FK constraint filter, continue without it
+ # This prevents the cleanup from failing, though it may still hit FK violations
Review Comment:
In the dag_version special-case, `base_table_id_col = base_table.c[dv_id_col.name]` will raise `KeyError` because `_TableConfig` for `dag_version` does not include the `id` column (it only declares `dag_id` + `created_at`). Since `KeyError` is caught, the FK-protection filter is silently skipped, so the fix won’t take effect. Add `id` to the dag_version config’s `extra_columns` (or build `base_table` from the reflected dag_version table when `table_name == "dag_version"`) so the correlated EXISTS can reference the PK column reliably.
```suggestion
base_table_id_col = base_table.c.get(dv_id_col.name)
if base_table_id_col is None:
# If the base_table does not expose the PK column (e.g. dag_version.id is missing
# from its configuration), we cannot safely build the FK-protection filter here.
# Log explicitly and continue without adding the EXISTS condition.
logger.warning(
"dag_version cleanup: base table has no '%s' column; "
"skipping foreign key protection filter against task_instance.dag_version_id.",
dv_id_col.name,
)
else:
kept_tis_exists = exists(
select(1)
.select_from(ti_table_reflected)
.where(ti_dag_version_id_col == base_table_id_col)
.where(ti_start_date_col >= clean_before_timestamp)
.where(ti_dag_version_id_col.isnot(None))
)
# Exclude dag_version rows that are referenced by kept task_instance rows
# Negate EXISTS to get NOT EXISTS behavior
conditions.append(~kept_tis_exists)
except (OperationalError, ProgrammingError) as e:
# If we can't add the FK constraint filter due to database/SQL issues, continue without it.
# This prevents the cleanup from failing, though it may still hit FK violations.
```
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -741,6 +741,318 @@ def test_drop_archived_tables(self, mock_input, confirm_mock, inspect_mock, capl
else:
confirm_mock.assert_not_called()
+ def test_dag_version_cleanup_respects_task_instance_fk_constraint(self):
+ """
+ Verify that dag_version cleanup does not delete dag_version rows that are
+ referenced by task_instance rows that are being kept (not deleted).
+
+ This tests the FK constraint handling added to prevent:
+ - task_instance.dag_version_id uses ondelete="RESTRICT", so deleting a
+ dag_version that's still referenced by a kept task_instance would fail.
+
+ The test creates:
+ - 2 dag_versions (old and new)
+ - 2 task_instances: one old (to be deleted) and one new (to be kept)
+ - Both TIs reference different dag_versions
+
+ When cleaning up with a timestamp that keeps the new TI, the dag_version
+ referenced by the kept TI should NOT be deleted, even if it's old.
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
+
+ with create_session() as session:
+ bundle_name = "testing"
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag_id = "test-dag-fk-constraint"
+ dag = DAG(dag_id=dag_id)
+ dm = DagModel(dag_id=dag_id, bundle_name=bundle_name)
+ session.add(dm)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=bundle_name)
+ dag_version = DagVersion.get_latest_version(dag.dag_id)
+
+ # Create an old dag_run and task_instance (to be deleted)
+ old_start_date = base_date
+ old_dag_run = DagRun(
+ dag.dag_id,
+ run_id="old_run",
+ run_type=DagRunType.MANUAL,
+ start_date=old_start_date,
+ )
+ old_ti = create_task_instance(
+ PythonOperator(task_id="old-task", python_callable=print),
+ run_id=old_dag_run.run_id,
+ dag_version_id=dag_version.id,
+ )
+ old_ti.dag_id = dag.dag_id
+ old_ti.start_date = old_start_date
+ session.add(old_dag_run)
+ session.add(old_ti)
+
+ # Create a new dag_run and task_instance (to be kept)
+ new_start_date = base_date.add(days=10)
+ new_dag_run = DagRun(
+ dag.dag_id,
+ run_id="new_run",
+ run_type=DagRunType.MANUAL,
+ start_date=new_start_date,
+ )
+ new_ti = create_task_instance(
+ PythonOperator(task_id="new-task", python_callable=print),
+ run_id=new_dag_run.run_id,
+ dag_version_id=dag_version.id,
Review Comment:
The docstring says the two task_instances reference different dag_versions, but both `create_task_instance(...)` calls pass `dag_version_id=dag_version.id`. If the intent is to model the FK-violation scenario, the kept task_instance should reference an *older* dag_version that would otherwise be deleted (i.e., not the keep_last version).
```suggestion
# Use an older dag_version for the kept task_instance to model the FK-violation scenario.
older_dag_version = session.scalars(
select(DagVersion)
.where(DagVersion.dag_id == dag_id)
.order_by(DagVersion.version_number)
).first()
new_ti = create_task_instance(
PythonOperator(task_id="new-task", python_callable=print),
run_id=new_dag_run.run_id,
dag_version_id=older_dag_version.id if older_dag_version is not None else dag_version.id,
```
--
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