You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "onlyarnav (via GitHub)" <gi...@apache.org> on 2026/06/30 11:13:25 UTC
[PR] Fix airflow db clean failure with DAG filtering on joined tables [airflow]
onlyarnav opened a new pull request, #69177:
URL: https://github.com/apache/airflow/pull/69177
Fix airflow db clean failure with DAG filtering on joined tables
closes: #69030
### Description
When running `airflow db clean` with `--dag-ids` or `--exclude-dag-ids`, the command previously failed with a `RuntimeError` on the `asset_event`, `deadline`, and `task_reschedule` tables.
This occurred because the `_TableConfig` for these tables incorrectly declared `dag_id_column_name="dag_id"`, but these tables either do not have a `dag_id` column or require a JOIN to resolve it.
**Changes made in this PR:**
* **`asset_event`**: Mapped the correct column by changing it to `dag_id_column_name="source_dag_id"`.
* **`deadline`**: Removed the `dag_id` column mapping and added `dag_id_join_info` to JOIN the `dag_run` table and filter by `dag_run.dag_id`.
* **`task_reschedule`**: Removed the `dag_id` column mapping and added `dag_id_join_info` to JOIN the `task_instance` table and filter by `task_instance.dag_id`.
* Added a new unit test `test_cleanup_with_dag_id_filtering_for_joined_tables` to explicitly verify the SQL generation and JOIN logic for these tables.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
`Claude Code (Opus 4.8)`
--
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 airflow db clean failure with DAG filtering on joined tables [airflow]
Posted by "onlyarnav (via GitHub)" <gi...@apache.org>.
onlyarnav closed pull request #69177: Fix airflow db clean failure with DAG filtering on joined tables
URL: https://github.com/apache/airflow/pull/69177
--
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 airflow db clean failure with DAG filtering on joined tables [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69177:
URL: https://github.com/apache/airflow/pull/69177#discussion_r3498651236
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -443,6 +443,112 @@ 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_cleanup_with_dag_id_filtering_for_joined_tables(self):
+ """
+ Verify that dag_ids and exclude_dag_ids parameters correctly filter and cleanup
+ asset_event, task_reschedule, and deadline tables.
+ """
+ from airflow.models.asset import AssetEvent
+ from airflow.models.deadline import Deadline
+ from airflow.models.taskreschedule import TaskReschedule
+
+ class SyncCallback:
+ def __init__(self):
+ self.path = "my_path"
+ self.kwargs = {}
+ self.executor = "default"
+
+ def serialize(self):
+ return {"path": self.path, "kwargs": self.kwargs, "executor": self.executor}
+
+ 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 records for dag1 (should be cleaned) and dag2 (should be kept)
+ for dag_id in ["dag1", "dag2"]:
+ 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)
+
+ dag_run = DagRun(
+ dag.dag_id,
+ run_id=f"{dag_id}_run",
+ run_type=DagRunType.MANUAL,
+ start_date=base_date,
+ )
+ session.add(dag_run)
+ session.flush()
+
+ 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 = base_date
+ session.add(ti)
+ session.flush()
+
+ # 1. AssetEvent
+ ae = AssetEvent(
+ asset_id=1,
+ source_dag_id=dag_id,
+ timestamp=base_date,
+ )
+ session.add(ae)
+
+ # 2. TaskReschedule
+ tr = TaskReschedule(
+ ti_id=ti.id,
+ start_date=base_date,
+ end_date=base_date.add(minutes=5),
+ reschedule_date=base_date.add(minutes=10),
+ )
+ session.add(tr)
+
+ # 3. Deadline
+ dl = Deadline(
+ deadline_time=base_date,
+ callback=SyncCallback(),
+ dagrun_id=dag_run.id,
+ deadline_alert_id=None,
+ )
+ session.add(dl)
+
+ session.commit()
+
+ # Run cleanup targeting only dag1
+ clean_before_date = base_date.add(days=10)
+ run_cleanup(
+ clean_before_timestamp=clean_before_date,
+ table_names=["asset_event", "task_reschedule", "deadline"],
Review Comment:
The test passes only the three joined tables, so it validates the JOIN-resolves-dag_id logic but skips the path that actually runs in airflow db clean: a full run cleans `task_instance`/`dag_run` first, which cascade-deletes the child rows before the child's own dag-filter ever fires.
Worth a second test case (or extending this one) that includes `task_instance` and `dag_run` in table_names so the cascade-interaction path is covered
--
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 airflow db clean failure with DAG filtering on joined tables [airflow]
Posted by "onlyarnav (via GitHub)" <gi...@apache.org>.
onlyarnav commented on code in PR #69177:
URL: https://github.com/apache/airflow/pull/69177#discussion_r3499396051
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -443,6 +443,112 @@ 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_cleanup_with_dag_id_filtering_for_joined_tables(self):
+ """
+ Verify that dag_ids and exclude_dag_ids parameters correctly filter and cleanup
+ asset_event, task_reschedule, and deadline tables.
+ """
+ from airflow.models.asset import AssetEvent
+ from airflow.models.deadline import Deadline
+ from airflow.models.taskreschedule import TaskReschedule
+
+ class SyncCallback:
+ def __init__(self):
+ self.path = "my_path"
+ self.kwargs = {}
+ self.executor = "default"
+
+ def serialize(self):
+ return {"path": self.path, "kwargs": self.kwargs, "executor": self.executor}
+
+ 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 records for dag1 (should be cleaned) and dag2 (should be kept)
+ for dag_id in ["dag1", "dag2"]:
+ 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)
+
+ dag_run = DagRun(
+ dag.dag_id,
+ run_id=f"{dag_id}_run",
+ run_type=DagRunType.MANUAL,
+ start_date=base_date,
+ )
+ session.add(dag_run)
+ session.flush()
+
+ 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 = base_date
+ session.add(ti)
+ session.flush()
+
+ # 1. AssetEvent
+ ae = AssetEvent(
+ asset_id=1,
+ source_dag_id=dag_id,
+ timestamp=base_date,
+ )
+ session.add(ae)
+
+ # 2. TaskReschedule
+ tr = TaskReschedule(
+ ti_id=ti.id,
+ start_date=base_date,
+ end_date=base_date.add(minutes=5),
+ reschedule_date=base_date.add(minutes=10),
+ )
+ session.add(tr)
+
+ # 3. Deadline
+ dl = Deadline(
+ deadline_time=base_date,
+ callback=SyncCallback(),
+ dagrun_id=dag_run.id,
+ deadline_alert_id=None,
+ )
+ session.add(dl)
+
+ session.commit()
+
+ # Run cleanup targeting only dag1
+ clean_before_date = base_date.add(days=10)
+ run_cleanup(
+ clean_before_timestamp=clean_before_date,
+ table_names=["asset_event", "task_reschedule", "deadline"],
Review Comment:
Thanks for the suggestion. I have added a `@pytest.mark.parametrize` which now runs the same test function twice.
```
first: ["asset_event", "task_reschedule", "deadline"] (the original logic).
second: ["asset_event", "task_reschedule", "deadline", "task_instance", "dag_run"]
```
Therefor covering both the paths.
--
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