remove stale serialized dags

[pingzh]

the processor should clean up stale serialized dags after parsing a dag file. The staleness comes from deleting a dag, renaming dag_id in a dag file.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

[pingzh]

@ashb could you please take a look when you have chance? thanks

[potiuk]

I think that one should wait for 2.4.0dev - adding more to 2.3.0 now unless it's really necessary is a bit risky :D

[pingzh]

@potiuk since 2.3 is released. can you please take a look at this PR?

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[SamWheating]

Thoughts on deleting missing DAGs from the serialized DAGs table at the same time that it is marked inactive?

airflow/airflow/dag_processing/manager.py

Lines 508 to 515 in cfa95af

	if to_deactivate:
	deactivated = (
	session.query(DagModel)
	.filter(DagModel.dag_id.in_(to_deactivate))
	.update({DagModel.is_active: False}, synchronize_session="fetch")
	)
	if deactivated:
	self.log.info("Deactivated %i DAGs which are no longer present in file.", deactivated)

There's some background in #21397 (Although I don't think the performance implications here are nearly as severe as in that case, as the serialized DAG table is properly indexed)

This query is gonna run really frequently (once for every DAG per parse interval) and most of the time won't have any missing DAGs to delete, by performing this on an intermittent basis rather than constantly, we can reduce the volume of queries.

This also would centralize similar pieces of code, rather than having similar functionality in completely separate parts of the codebase.

[pingzh]

Thoughts on deleting missing DAGs from the serialized DAGs table at the same time that it is marked inactive?

airflow/airflow/dag_processing/manager.py

Lines 508 to 515 in cfa95af

	if to_deactivate:
	deactivated = (
	session.query(DagModel)
	.filter(DagModel.dag_id.in_(to_deactivate))
	.update({DagModel.is_active: False}, synchronize_session="fetch")
	)
	if deactivated:
	self.log.info("Deactivated %i DAGs which are no longer present in file.", deactivated)

There's some background in #21397 (Although I don't think the performance implications here are nearly as severe as in that case, as the serialized DAG table is properly indexed)

This query is gonna run really frequently (once for every DAG per parse interval) and most of the time won't have any missing DAGs to delete, by performing this on an intermittent basis rather than constantly, we can reduce the volume of queries.

This also would centralize similar pieces of code, rather than having similar functionality in completely separate parts of the codebase.

@SamWheating thanks for the context. that is a good idea to keep them together. updating the pr.

[KevinYang21]

Thanks everyone for help reviewing the PR. Given the PR has been open for a while and all comments have been addressed, if no more comments, I'll merge this PR tonight.

[potiuk]

Thanks everyone for help reviewing the PR. Given the PR has been open for a while and all comments have been addressed, if no more comments, I'll merge this PR tonight.

Just released the providers RC (with some 'surprises') - I will take a look tomorrow finally if you can wait.