[v3-3-test] Improve deadline diagnostics for null DagRun fields (#71767)

airflow-core/newsfragments/71767.bugfix.rst

@@ -0,0 +1 @@
+Fix misleading deadline warnings when DagRun-based deadline references evaluate to None because the referenced timestamp field is null.

airflow-core/src/airflow/serialization/definitions/dag.py

@@ -82,6 +82,11 @@
 
 log = structlog.get_logger(__name__)
 
+_DAGRUN_REFERENCE_REQUIRED_COLUMNS = {
+    SerializedReferenceModels.DagRunLogicalDateDeadline: "logical_date",
+    SerializedReferenceModels.DagRunQueuedAtDeadline: "queued_at",
+}
+
 
 # TODO (GH-52141): Share definition with SDK?
 class EdgeInfoType(TypedDict):
@@ -763,6 +768,17 @@ def _process_dagrun_deadline_alerts(
                         "deadline_alerts.deadline_created",
                         tags=prune_dict({"dag_id": self.dag_id, "team_name": team_name}),
                     )
+                elif required_dagrun_column := _DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
+                    type(deserialized_deadline_alert.reference)
+                ):
+                    log.warning(
+                        "skipping deadline alert because the deadline reference evaluated to None",
+                        dag_id=self.dag_id,
+                        run_id=orm_dagrun.run_id,
+                        deadline_alert_id=deadline_alert.id,
+                        reference_type=deserialized_deadline_alert.reference.reference_name,
+                        required_dagrun_column=required_dagrun_column,
+                    )
 
     @provide_session
     def set_task_instance_state(

airflow-core/src/airflow/serialization/definitions/deadline.py

@@ -373,10 +373,13 @@ def _fetch_from_db(column, *, session: Session, dag_id: str, run_id: str) -> dat
     """
     from airflow.models import DagRun
 
-    result = session.execute(select(column).where(DagRun.dag_id == dag_id, DagRun.run_id == run_id)).scalar()
-    if result is None:
+    row = session.execute(
+        select(column).where(DagRun.dag_id == dag_id, DagRun.run_id == run_id)
+    ).one_or_none()
+    if row is None:
         logger.warning("Could not find DagRun for dag_id=%s, run_id=%s", dag_id, run_id)
-    return result
+        return None
+    return row[0]
 
 
 @attrs.define(frozen=True)

airflow-core/tests/unit/models/test_dagrun.py

@@ -1542,6 +1542,64 @@ def test_dagrun_deadline_variable_interval_stable(self, _, mock_get, session, de
         deadline = session.execute(select(Deadline)).scalars().one_or_none()
         assert deadline.deadline_time == first_deadline_time
 
+    def test_dagrun_deadline_logs_when_reference_column_is_null(self, session, deadline_test_dag, caplog):
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
+                interval=datetime.timedelta(minutes=5),
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        with caplog.at_level("WARNING"):
+            scheduler_dag.create_dagrun(
+                run_id="manual__null_logical_date",
+                run_type=DagRunType.MANUAL,
+                logical_date=None,
+                data_interval=None,
+                run_after=timezone.utcnow(),
+                start_date=timezone.utcnow(),
+                state=DagRunState.QUEUED,
+                triggered_by=DagRunTriggeredByType.TEST,
+                session=session,
+            )
+
+        assert session.execute(select(Deadline)).scalars().one_or_none() is None
+        assert {
+            "event": "skipping deadline alert because the deadline reference evaluated to None",
+            "dag_id": "test_dag",
+            "run_id": "manual__null_logical_date",
+            "reference_type": "DagRunLogicalDateDeadline",
+            "required_dagrun_column": "logical_date",
+            "log_level": "warning",
+        } in caplog
+        assert not any("Could not find DagRun" in record.message for record in caplog.records)
+
+    def test_dagrun_deadline_does_not_warn_for_average_runtime_without_history(
+        self, session, deadline_test_dag, caplog
+    ):
+        scheduler_dag = deadline_test_dag(
+            deadline=DeadlineAlert(
+                reference=DeadlineReference.AVERAGE_RUNTIME(max_runs=10, min_runs=5),
+                interval=datetime.timedelta(minutes=5),
+                callback=AsyncCallback(empty_callback_for_deadline),
+            ),
+        )
+
+        with caplog.at_level("WARNING", logger="airflow.serialization.definitions.dag"):
+            self.create_dag_run(
+                dag=scheduler_dag,
+                logical_date=DEFAULT_DATE,
+                session=session,
+            )
+
+        assert session.execute(select(Deadline)).scalars().one_or_none() is None
+        assert {
+            "event": "skipping deadline alert because the deadline reference evaluated to None",
+            "reference_type": "AverageRuntimeDeadline",
+            "log_level": "warning",
+        } not in caplog
+
     @mock.patch.object(Deadline, "prune_deadlines")
     def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, session, deadline_test_dag):
         mock_err = mock.Mock()

airflow-core/tests/unit/models/test_deadline.py

@@ -42,7 +42,10 @@
     FixedDatetimeDeadline,
     deadline_reference,
 )
-from airflow.serialization.definitions.deadline import SerializedReferenceModels
+from airflow.serialization.definitions.deadline import (
+    SerializedReferenceModels,
+    _fetch_from_db as _fetch_serialized_deadline_from_db,
+)
 from airflow.utils.state import DagRunState
 
 from tests_common.test_utils import db
@@ -476,6 +479,36 @@ def test_deadline_database_integration(self, reference, expected_column, session
                 mock_fetch.assert_not_called()
                 assert result == DEFAULT_DATE + interval
 
+    def test_serialized_fetch_from_db_returns_null_column_without_missing_dagrun_warning(
+        self, session, dag_maker, caplog
+    ):
+        with dag_maker(DAG_ID):
+            EmptyOperator(task_id="test_task")
+
+        dag_maker.create_dagrun(
+            logical_date=None,
+            run_id="manual__null_logical_date",
+            state=DagRunState.QUEUED,
+        )
+        session.commit()
+
+        with caplog.at_level("WARNING", logger="airflow.serialization.definitions.deadline"):
+            result = _fetch_serialized_deadline_from_db(
+                DagRun.logical_date, session=session, dag_id=DAG_ID, run_id="manual__null_logical_date"
+            )
+
+        assert result is None
+        assert not any("Could not find DagRun" in record.message for record in caplog.records)
+
+    def test_serialized_fetch_from_db_logs_missing_dagrun(self, session, caplog):
+        with caplog.at_level("WARNING", logger="airflow.serialization.definitions.deadline"):
+            result = _fetch_serialized_deadline_from_db(
+                DagRun.logical_date, session=session, dag_id=DAG_ID, run_id="missing_run"
+            )
+
+        assert result is None
+        assert any("Could not find DagRun" in record.message for record in caplog.records)
+
     def test_average_runtime_with_sufficient_history(self, session, dag_maker):
         """Test AverageRuntimeDeadline when enough historical data exists."""
         with dag_maker(DAG_ID):