fix(datasets): demote missing SerializedDagModel DDRQ log to debug

Log the DDRQ-without-serialization case at debug and remove the
[DEBUG DATASETS] prefix; drop redundant del of missing_from_serialized.
Tests capture DEBUG, match the new message, and assert
dataset_dag_run_queue rows remain after dags_needing_dagruns.

Author: leossantos

airflow/models/dag.py

@@ -4097,15 +4097,14 @@ def dag_ready(dag_id: str, cond: BaseDataset, statuses: dict) -> bool | None:
         ser_dag_ids = {s.dag_id for s in ser_dags}
         missing_from_serialized = set(by_dag.keys()) - ser_dag_ids
         if missing_from_serialized:
-            log.warning(
-                "[DEBUG DATASETS] DAGs in DDRQ but missing SerializedDagModel "
+            log.debug(
+                "DAGs in DDRQ but missing SerializedDagModel "
                 "(skipping — condition cannot be evaluated): %s",
                 sorted(missing_from_serialized),
             )
             for dag_id in missing_from_serialized:
                 del by_dag[dag_id]
                 del dag_statuses[dag_id]
-        del missing_from_serialized
         for ser_dag in ser_dags:
             dag_id = ser_dag.dag_id
             statuses = dag_statuses[dag_id]

tests/models/test_dag.py

@@ -3065,7 +3065,12 @@ def test_dags_needing_dagruns_datasets(self, dag_maker, session):
         assert dag_models == [dag_model]
 
     def test_dags_needing_dagruns_skips_ddrq_when_serialized_dag_missing(self, session, caplog):
-        """DDRQ rows for a dag_id without SerializedDagModel must be skipped (no dataset_triggered info)."""
+        """DDRQ rows for a dag_id without SerializedDagModel must be skipped (no dataset_triggered info).
+
+        Rows must remain in ``dataset_dag_run_queue`` so the scheduler can re-evaluate on a later
+        heartbeat once ``SerializedDagModel`` exists (``dags_needing_dagruns`` only drops them from
+        the in-memory candidate set, it does not delete ORM rows).
+        """
         orphan_dag_id = "ddr_q_no_serialized_dag"
         session.add(DatasetModel(uri="dataset_for_orphan_ddrq"))
         session.flush()
@@ -3085,12 +3090,18 @@ def test_dags_needing_dagruns_skips_ddrq_when_serialized_dag_missing(self, sessi
         session.add(DatasetDagRunQueue(dataset_id=dataset_id, target_dag_id=orphan_dag_id))
         session.flush()
 
-        with caplog.at_level(logging.WARNING, logger="airflow.models.dag"):
+        with caplog.at_level(logging.DEBUG, logger="airflow.models.dag"):
             _query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
 
         assert orphan_dag_id not in dataset_triggered_dag_info
-        assert "[DEBUG DATASETS] DAGs in DDRQ but missing SerializedDagModel" in caplog.text
+        assert "DAGs in DDRQ but missing SerializedDagModel" in caplog.text
         assert orphan_dag_id in caplog.text
+        assert (
+            session.query(DatasetDagRunQueue)
+            .filter(DatasetDagRunQueue.target_dag_id == orphan_dag_id)
+            .count()
+            == 1
+        )
 
     def test_dags_needing_dagruns_missing_serialized_warning_lists_sorted_dag_ids(self, session, caplog):
         """When multiple dags lack SerializedDagModel, the warning lists dag_ids sorted."""
@@ -3133,17 +3144,21 @@ def test_dags_needing_dagruns_missing_serialized_warning_lists_sorted_dag_ids(se
         )
         session.flush()
 
-        with caplog.at_level(logging.WARNING, logger="airflow.models.dag"):
+        with caplog.at_level(logging.DEBUG, logger="airflow.models.dag"):
             _query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
 
         assert "ghost_a" not in dataset_triggered_dag_info
         assert "ghost_z" not in dataset_triggered_dag_info
         msg = next(
-            r.message
-            for r in caplog.records
-            if "[DEBUG DATASETS] DAGs in DDRQ but missing SerializedDagModel" in r.message
+            r.message for r in caplog.records if "DAGs in DDRQ but missing SerializedDagModel" in r.message
         )
         assert msg.index("ghost_a") < msg.index("ghost_z")
+        assert (
+            session.query(DatasetDagRunQueue)
+            .filter(DatasetDagRunQueue.target_dag_id.in_(("ghost_a", "ghost_z")))
+            .count()
+            == 2
+        )
 
     def test_dags_needing_dagruns_dataset_aliases(self, dag_maker, session):
         # link dataset_alias hello_alias to dataset hello