You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/04/10 19:59:58 UTC
Re: [PR] fix: skip dag_version rows still referenced by task_instance in db clean [airflow]
Copilot commented on code in PR #63728:
URL: https://github.com/apache/airflow/pull/63728#discussion_r3066477950
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -422,6 +422,65 @@ def test_cleanup_with_dag_id_filtering(self, dag_ids, exclude_dag_ids, expected_
f"Expected {expected_remaining_dag_ids} to remain, but got {remaining_dag_ids}"
)
+ def test_dag_version_cleanup_skips_referenced_rows(self):
+ """
+ Verify that dag_version rows still referenced by task_instance are not deleted,
+ preventing ForeignKeyViolation (issue #63703).
+
+ Scenario: dag_version created before the cutoff, but a task_instance created
+ after the cutoff still references it via dag_version_id (RESTRICT FK).
+ """
+ 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-version-fk"
+ 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 intends `dag_version.created_at` to be before `clean_before_date`, but `created_at` is never set/frozen—so it will typically be ‘now’ and therefore *after* the cutoff, meaning the row wouldn’t be eligible for deletion in the first place. To ensure the test actually reproduces the FK-violation scenario, freeze time for the dag_version creation or explicitly set `dag_version.created_at` to `base_date` (and assert the precondition `dag_version.created_at < clean_before_date`).
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -422,6 +422,65 @@ def test_cleanup_with_dag_id_filtering(self, dag_ids, exclude_dag_ids, expected_
f"Expected {expected_remaining_dag_ids} to remain, but got {remaining_dag_ids}"
)
+ def test_dag_version_cleanup_skips_referenced_rows(self):
+ """
+ Verify that dag_version rows still referenced by task_instance are not deleted,
+ preventing ForeignKeyViolation (issue #63703).
+
+ Scenario: dag_version created before the cutoff, but a task_instance created
+ after the cutoff still references it via dag_version_id (RESTRICT FK).
+ """
+ base_date = pendulum.DateTime(2022, 1, 1, tzinfo=pendulum.timezone("UTC"))
Review Comment:
This test intends `dag_version.created_at` to be before `clean_before_date`, but `created_at` is never set/frozen—so it will typically be ‘now’ and therefore *after* the cutoff, meaning the row wouldn’t be eligible for deletion in the first place. To ensure the test actually reproduces the FK-violation scenario, freeze time for the dag_version creation or explicitly set `dag_version.created_at` to `base_date` (and assert the precondition `dag_version.created_at < clean_before_date`).
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -422,6 +422,65 @@ def test_cleanup_with_dag_id_filtering(self, dag_ids, exclude_dag_ids, expected_
f"Expected {expected_remaining_dag_ids} to remain, but got {remaining_dag_ids}"
)
+ def test_dag_version_cleanup_skips_referenced_rows(self):
+ """
+ Verify that dag_version rows still referenced by task_instance are not deleted,
+ preventing ForeignKeyViolation (issue #63703).
+
+ Scenario: dag_version created before the cutoff, but a task_instance created
+ after the cutoff still references it via dag_version_id (RESTRICT FK).
+ """
+ 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-version-fk"
+ 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 a dag_run and task_instance with start_date AFTER the cutoff,
+ # but referencing the old dag_version (created before cutoff).
+ run_start = base_date.add(days=20)
+ dag_run = DagRun(
+ dag.dag_id,
+ run_id="run_after_cutoff",
+ run_type=DagRunType.MANUAL,
+ start_date=run_start,
+ )
+ ti = create_task_instance(
+ PythonOperator(task_id="dummy-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 = run_start
+ session.add(dag_run)
+ session.add(ti)
+ session.commit()
+
+ # Clean with a cutoff that is after the dag_version creation but before
+ # the task_instance start_date.
+ clean_before_date = base_date.add(days=10)
Review Comment:
This test intends `dag_version.created_at` to be before `clean_before_date`, but `created_at` is never set/frozen—so it will typically be ‘now’ and therefore *after* the cutoff, meaning the row wouldn’t be eligible for deletion in the first place. To ensure the test actually reproduces the FK-violation scenario, freeze time for the dag_version creation or explicitly set `dag_version.created_at` to `base_date` (and assert the precondition `dag_version.created_at < clean_before_date`).
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -373,6 +382,21 @@ def _build_query(
)
conditions.append(column(max_date_col_name).is_(None))
query = query.where(and_(*conditions))
+
+ # Exclude rows still referenced by child tables (FK with RESTRICT).
+ fk_check_columns = kwargs.get("fk_check_columns")
+ if fk_check_columns:
+ for child_table_name, child_fk_col, parent_pk_col in fk_check_columns:
+ child_table = table(child_table_name, column(child_fk_col))
+ parent_pk = getattr(base_table, parent_pk_col)
+ query = query.where(
+ ~exists(
+ select(literal(1))
+ .select_from(child_table)
+ .where(child_table.c[child_fk_col] == parent_pk)
+ )
+ )
Review Comment:
`base_table` is an aliased selectable and columns are accessed via `.c[...]` elsewhere in this function. Using `getattr(base_table, parent_pk_col)` is likely to fail (or produce the wrong SQL expression) because `id`/PK columns generally aren’t exposed as attributes on these alias objects. Use `base_table.c[parent_pk_col]` (and consider raising a clear error if the PK column isn’t present in the selectable) to ensure the FK check is actually correlated correctly.
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -373,6 +382,21 @@ def _build_query(
)
conditions.append(column(max_date_col_name).is_(None))
query = query.where(and_(*conditions))
+
+ # Exclude rows still referenced by child tables (FK with RESTRICT).
+ fk_check_columns = kwargs.get("fk_check_columns")
+ if fk_check_columns:
Review Comment:
`fk_check_columns` is now a first-class behavior change in `_build_query`, but it’s threaded via `**kwargs`, which makes the API harder to understand and type-check. Consider adding an explicit, typed keyword parameter (e.g. `fk_check_columns: list[tuple[str, str, str]] | None = None`) to `_build_query` (and similarly to `_cleanup_table` if needed) rather than relying on `kwargs.get`.
```suggestion
if fk_check_columns is not None:
```
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -88,6 +88,11 @@ class _TableConfig:
# because the relationships are unlikely to change and the number of tables is small.
# Relying on automation here would increase complexity and reduce maintainability.
dependent_tables: list[str] | None = None
+ # Columns in other tables that hold FK references to this table's primary key.
+ # Used to add NOT EXISTS guards so we don't delete rows still referenced by
+ # surviving child rows (e.g. task_instance.dag_version_id → dag_version.id).
+ # Each entry is (child_table_name, child_fk_column_name, parent_pk_column_name).
+ fk_check_columns: list[tuple[str, str, str]] | None = None
Review Comment:
Since `fk_check_columns` changes which rows are eligible for deletion, it would be helpful for `dry_run`/config output to surface it (similar to `dependent_tables` and `keep_last*` outputs). Consider including `fk_check_columns` in the human-readable config that gets printed so operators can understand why some parent rows are being skipped.
--
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