Apply suggestions from code review

Co-authored-by: Wei Lee <weilee.rx@gmail.com>

Author: leossantos

airflow/models/dag.py

@@ -4070,9 +4070,9 @@ def dags_needing_dagruns(cls, session: Session) -> tuple[Query, dict[str, tuple[
         you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
         transaction is committed it will be unlocked.
 
-        For dataset-triggered scheduling, DAGs that have ``DatasetDagRunQueue`` rows but no matching
+        For dataset-triggered scheduling, Dags that have ``DatasetDagRunQueue`` rows but no matching
         ``SerializedDagModel`` row are omitted from the returned ``dataset_triggered_dag_info`` until
-        serialization exists; queue rows are **not** deleted here so the scheduler can re-evaluate on a
+        serialization exists; DDRQs are **not** deleted here so the scheduler can re-evaluate on a
         later run.
         """
         from airflow.models.serialized_dag import SerializedDagModel
@@ -4100,16 +4100,16 @@ def dag_ready(dag_id: str, cond: BaseDataset, statuses: dict) -> bool | None:
             select(SerializedDagModel).where(SerializedDagModel.dag_id.in_(dag_statuses.keys()))
         ).all()
         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.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]
+        missing_from_serialized = set(by_dag.keys()) - ser_dag_ids
+        if missing_from_serialized:
+            log.debug(
+                "Dags have queued dataset events (DDRQs), but are not found in the serialized_dag table."
+                " — skipping Dag run creation: %s",
+                sorted(missing_from_serialized),
+            )
+            for dag_id in missing_from_serialized:
+                del by_dag[dag_id]
+                del dag_statuses[dag_id]
         for ser_dag in ser_dags:
             dag_id = ser_dag.dag_id
             statuses = dag_statuses[dag_id]

newsfragments/63546.bugfix.rst

@@ -1 +1 @@
-Fix premature dataset-triggered DagRuns when ``SerializedDagModel`` was missing while ``DatasetDagRunQueue`` still had rows for that DAG; queue entries are kept for the next evaluation.
+Fix premature dataset-triggered DagRuns when ``SerializedDagModel`` was missing while ``DatasetDagRunQueue`` still had rows for that Dag; queue entries are kept for the next evaluation.

tests/models/test_dag.py

@@ -3066,7 +3066,7 @@ 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 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