[AIRFLOW-842] do not query the DB with an empty IN clause

This is done to silence warnings coming from
sqlachemy, e.g.:

sqlalchemy/sql/default_comparator.py:161:
SAWarning: The IN-predicate on
"dag_run.dag_id" was invoked with an empty
sequence. This results in a
contradiction, which nonetheless can be expensive
to evaluate. Consider
alternative strategies for improved performance.

Closes apache#2072 from imbaczek/bug842

airflow/models.py

@@ -3734,6 +3734,10 @@ def clean_dirty(dag_ids, session=None):
         :param full_query: whether to check dag_runs for new drs not in dag_stats
         :type full_query: bool
         """
+        # avoid querying with an empty IN clause
+        if not dag_ids:
+            return
+
         dag_ids = set(dag_ids)
 
         qry = (
@@ -3745,6 +3749,10 @@ def clean_dirty(dag_ids, session=None):
         qry.delete(synchronize_session='fetch')
         session.commit()
 
+        # avoid querying with an empty IN clause
+        if not dirty_ids:
+            return
+
         qry = (
             session.query(DagRun.dag_id, DagRun.state, func.count('*'))
             .filter(DagRun.dag_id.in_(dirty_ids))

tests/core.py

@@ -978,12 +978,18 @@ def test_dag_stats(self):
         session.query(models.DagStat).delete()
         session.commit()
 
+        with warnings.catch_warnings(record=True) as caught_warnings:
+            models.DagStat.clean_dirty([], session=session)
+        self.assertEqual([], caught_warnings)
+
         run1 = self.dag_bash.create_dagrun(
             run_id="run1",
             execution_date=DEFAULT_DATE,
             state=State.RUNNING)
 
-        models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        with warnings.catch_warnings(record=True) as caught_warnings:
+            models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        self.assertEqual([], caught_warnings)
 
         qry = session.query(models.DagStat).all()
 
@@ -998,7 +1004,9 @@ def test_dag_stats(self):
             execution_date=DEFAULT_DATE+timedelta(days=1),
             state=State.RUNNING)
 
-        models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        with warnings.catch_warnings(record=True) as caught_warnings:
+            models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        self.assertEqual([], caught_warnings)
 
         qry = session.query(models.DagStat).all()
 
@@ -1011,7 +1019,9 @@ def test_dag_stats(self):
         session.query(models.DagRun).first().state = State.SUCCESS
         session.commit()
 
-        models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        with warnings.catch_warnings(record=True) as caught_warnings:
+            models.DagStat.clean_dirty([self.dag_bash.dag_id], session=session)
+        self.assertEqual([], caught_warnings)
 
         qry = session.query(models.DagStat).filter(models.DagStat.state == State.SUCCESS).all()
         self.assertEqual(1, len(qry))