jedcunningham commented on code in PR #27828:
URL: https://github.com/apache/airflow/pull/27828#discussion_r1032471565
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1574,3 +1585,33 @@ def _cleanup_stale_dags(self, session: Session =
NEW_SESSION) -> None:
dag.is_active = False
SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
session.flush()
+
+ @provide_session
+ def _orphan_unreferenced_datasets(self, session: Session = NEW_SESSION) ->
None:
+ """
+ Detects datasets that are no longer referenced in any DAG schedule
parameters or task outlets and
+ sets the dataset is_orphaned flag to True
+ """
+ orphaned_dataset_query = (
+ session.query(DatasetModel)
+ .join(
+ DagScheduleDatasetReference,
+ isouter=True,
+ )
+ .join(
+ TaskOutletDatasetReference,
+ isouter=True,
+ )
+ # MSSQL doesn't like it when we select a column that we haven't
grouped by. All other DBs let us
+ # group by id and select all columns.
+ .group_by(DatasetModel if session.get_bind().dialect.name ==
"mssql" else DatasetModel.id)
Review Comment:
Works for me. We can always tweak that later too.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
