blag commented on code in PR #27828:
URL: https://github.com/apache/airflow/pull/27828#discussion_r1032465764
##########
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:
The idealist in me thinks this is the Proper Way ™️ do do it, it's working,
and I still think it's pretty clear what's going on (read: what SQL is being
emitted).
The pragmatist in me recognizes that you're making a good point, but I think
let's leave it unless or until it causes issues, then do that at that point.
How does that plan sound?
--
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]
