You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@airflow.apache.org by "ferruzzi (via GitHub)" <gi...@apache.org> on 2026/02/10 04:44:25 UTC

[PR] Fix deadline alert hashing bug [airflow]

ferruzzi opened a new pull request, #61702:
URL: https://github.com/apache/airflow/pull/61702

   When I implemented UUID reuse to reduce DAG re-serialization, my strategy was too overzealous.  I blindly reused UUIDs if the DAG previously had ANY deadlines, which caused two bugs:
   
   1. If the Dag had multiple DeadlineAlerts, they failed silently (no new DeadlineAlert records were created)
   2. Deadline property changes (interval/reference/callback) were ignored if you edited the Dag once it was initially parsed
   
   We now compare existing DeadlineAlert records and check if any definitions changed.
   
   **New logic:**
   - Reuse UUIDs when ALL deadlines match (preserves hash, avoids unnecessary writes and re-serialization)
   - Generate new UUIDs when any deadline changes (updates the hash and creates new records)
   
   Also improved the relevant unit tests which would have caught this to prevent regression in the future.
   
   <!--
   Thank you for contributing!
   
   Please provide above a brief description of the changes made in this pull request.
   Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
   
   Please make sure that your code changes are covered with tests.
   And in case of new features or big changes remember to adjust the documentation.
   
   Feel free to ping (in general) for the review if you do not see reaction for a few days
   (72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
   
   In case of an existing issue, reference it using one of the following:
   
   * closes: #ISSUE
   * related: #ISSUE
   -->
   
   ---
   
   ##### Was generative AI tooling used to co-author this PR?
   
   <!--
   If generative AI tooling has been used in the process of authoring this PR, please
   change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
   -->
   
   - [ ] Yes (please specify the tool below)
   
   <!--
   Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
   -->
   
   ---
   
   * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
   * For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
   * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
   * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deadline alert hashing bug [airflow]

Posted by "Lee-W (via GitHub)" <gi...@apache.org>.

Lee-W commented on code in PR #61702:
URL: https://github.com/apache/airflow/pull/61702#discussion_r2790986234


##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -418,6 +418,68 @@ def _generate_deadline_uuids(cls, dag_data: dict[str, Any]) -> dict[str, dict]:
 
         return uuid_mapping
 
+    @classmethod
+    def _try_reuse_deadline_uuids(
+        cls,
+        existing_deadline_uuids: list[str],
+        new_deadline_data: list[dict],
+        session: Session,
+    ) -> dict[str, dict] | None:
+        """
+        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
+
+        Returns None if Deadline hashes are not all identical, indicating they need to be updated.
+
+        :param existing_deadline_uuids: List of UUID strings from existing serialized dag

Review Comment:
   non-blocking. I just point out when I noticed :)



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deadline alert hashing bug [airflow]

Posted by "vincbeck (via GitHub)" <gi...@apache.org>.

vincbeck commented on code in PR #61702:
URL: https://github.com/apache/airflow/pull/61702#discussion_r2824364594


##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -418,6 +418,68 @@ def _generate_deadline_uuids(cls, dag_data: dict[str, Any]) -> dict[str, dict]:
 
         return uuid_mapping
 
+    @classmethod
+    def _try_reuse_deadline_uuids(
+        cls,
+        existing_deadline_uuids: list[str],
+        new_deadline_data: list[dict],
+        session: Session,
+    ) -> dict[str, dict] | None:
+        """
+        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
+
+        Returns None if Deadline hashes are not all identical, indicating they need to be updated.
+
+        :param existing_deadline_uuids: List of UUID strings from existing serialized dag
+        :param new_deadline_data: List of new deadline alert data dicts from the DAG
+        :param session: Database session
+        :return: UUID mapping dict if all match, None if any mismatch detected
+        """
+        if len(existing_deadline_uuids) != len(new_deadline_data):
+            return None
+
+        existing_alerts = session.scalars(
+            select(DeadlineAlertModel).where(DeadlineAlertModel.id.in_(existing_deadline_uuids))
+        ).all()
+
+        if len(existing_alerts) != len(existing_deadline_uuids):
+            return None
+
+        new_alerts_temp = []
+        for deadline_alert in new_deadline_data:
+            deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)
+            # Create a temporary alert for comparison
+            temp_alert = DeadlineAlertModel(
+                id="temp",  # id is required for the object but isn't included in the __eq__

Review Comment:
   I understand the confusion but I think, indeed, using anew function is better. Why? Because if you have 2 `DeadlineAlertModel` identical objects but `id` and/or `serialized_dag_id` and you do a `==` comparison between these 2 object, Python will say they are identical, which is wrong, they are not



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix deadline alert hashing bug [airflow]

Posted by "Lee-W (via GitHub)" <gi...@apache.org>.

Lee-W commented on code in PR #61702:
URL: https://github.com/apache/airflow/pull/61702#discussion_r2786064832


##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -418,6 +418,68 @@ def _generate_deadline_uuids(cls, dag_data: dict[str, Any]) -> dict[str, dict]:
 
         return uuid_mapping
 
+    @classmethod
+    def _try_reuse_deadline_uuids(
+        cls,
+        existing_deadline_uuids: list[str],
+        new_deadline_data: list[dict],
+        session: Session,
+    ) -> dict[str, dict] | None:
+        """
+        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
+
+        Returns None if Deadline hashes are not all identical, indicating they need to be updated.
+
+        :param existing_deadline_uuids: List of UUID strings from existing serialized dag
+        :param new_deadline_data: List of new deadline alert data dicts from the DAG

Review Comment:
   ```suggestion
           :param new_deadline_data: List of new deadline alert data dicts from the Dag
   ```



##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -418,6 +418,68 @@ def _generate_deadline_uuids(cls, dag_data: dict[str, Any]) -> dict[str, dict]:
 
         return uuid_mapping
 
+    @classmethod
+    def _try_reuse_deadline_uuids(
+        cls,
+        existing_deadline_uuids: list[str],
+        new_deadline_data: list[dict],
+        session: Session,
+    ) -> dict[str, dict] | None:
+        """
+        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
+
+        Returns None if Deadline hashes are not all identical, indicating they need to be updated.
+
+        :param existing_deadline_uuids: List of UUID strings from existing serialized dag
+        :param new_deadline_data: List of new deadline alert data dicts from the DAG
+        :param session: Database session
+        :return: UUID mapping dict if all match, None if any mismatch detected
+        """
+        if len(existing_deadline_uuids) != len(new_deadline_data):
+            return None
+
+        existing_alerts = session.scalars(
+            select(DeadlineAlertModel).where(DeadlineAlertModel.id.in_(existing_deadline_uuids))
+        ).all()
+
+        if len(existing_alerts) != len(existing_deadline_uuids):
+            return None
+
+        new_alerts_temp = []
+        for deadline_alert in new_deadline_data:
+            deadline_data = deadline_alert.get(Encoding.VAR, deadline_alert)
+            # Create a temporary alert for comparison
+            temp_alert = DeadlineAlertModel(
+                id="temp",  # id is required for the object but isn't included in the __eq__

Review Comment:
   Is it possible to have multiple DeadlineAlertModels with the same property but different IDs?
   
   DeadlineAlertModel(id=1, reference="the same"...)
   DeadlineAlertModel(id=2, reference="the same"...)
   
   If it's possible, it looks incorrect to use `__eq__` to me. Comparing those attributes directly might seem lengthy, but it is probably more accurate and less confusing.



##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -503,9 +565,20 @@ def write_dag(
                 and existing_serialized_dag.data
                 and (existing_deadline_uuids := existing_serialized_dag.data.get("dag", {}).get("deadline"))
             ):
-                dag.data["dag"]["deadline"] = existing_deadline_uuids
-                deadline_uuid_mapping = {}
+                deadline_uuid_mapping = cls._try_reuse_deadline_uuids(
+                    existing_deadline_uuids,
+                    dag.data["dag"]["deadline"],
+                    session,
+                )
+
+                if deadline_uuid_mapping is not None:
+                    # All deadlines matched, reuse the UUIDs to preserve hash.
+                    dag.data["dag"]["deadline"] = list(deadline_uuid_mapping.keys())
+                else:
+                    # At least one deadline has changed, generate new UUIDs and update the hash.
+                    deadline_uuid_mapping = cls._generate_deadline_uuids(dag.data)
             else:
+                # First time seeing this DAG with deadlines, generate new UUIDs and update the hash.

Review Comment:
   ```suggestion
                   # First time seeing this Dag with deadlines, generate new UUIDs and update the hash.
   ```



##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -418,6 +418,68 @@ def _generate_deadline_uuids(cls, dag_data: dict[str, Any]) -> dict[str, dict]:
 
         return uuid_mapping
 
+    @classmethod
+    def _try_reuse_deadline_uuids(
+        cls,
+        existing_deadline_uuids: list[str],
+        new_deadline_data: list[dict],
+        session: Session,
+    ) -> dict[str, dict] | None:
+        """
+        Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
+
+        Returns None if Deadline hashes are not all identical, indicating they need to be updated.
+
+        :param existing_deadline_uuids: List of UUID strings from existing serialized dag

Review Comment:
   ```suggestion
           :param existing_deadline_uuids: List of UUID strings from existing serialized Dag
   ```



##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1872,7 +1873,7 @@ def test_dagrun_deadline(self, reference_type, reference_column, testing_dag_bun
         assert dr.deadlines[0].deadline_time == getattr(dr, reference_column, DEFAULT_DATE) + interval
 
     def test_dag_with_multiple_deadlines(self, testing_dag_bundle, session):
-        """Test that a DAG with multiple deadlines stores all deadlines in the database."""
+        """Test that a DAG with multiple deadlines stores all deadlines and persists on re-serialization."""

Review Comment:
   ```suggestion
           """Test that a Dag with multiple deadlines stores all deadlines and persists on re-serialization."""
   ```



##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -1899,6 +1901,28 @@ def test_dag_with_multiple_deadlines(self, testing_dag_bundle, session):
 
         scheduler_dag = sync_dag_to_db(dag, session=session)
 
+        deadline_alerts = session.scalars(select(DeadlineAlertModel)).all()
+        assert len(deadline_alerts) == expected_num_deadlines
+        initial_uuids = {alert.id for alert in deadline_alerts}
+
+        # Re-serialize the DAG

Review Comment:
   ```suggestion
           # Re-serialize the Dag
   ```



##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -753,3 +763,48 @@ def test_write_dag_atomicity_on_dagcode_failure(self, dag_maker, session):
                 assert len(sdag.dag.task_dict) == 1, (
                     "SerializedDagModel should not be updated when write fails"
                 )
+
+    def test_deadline_interval_change_triggers_new_serdag(self, testing_dag_bundle, session):
+        dag_id = "test_interval_change"
+
+        # Create a new Dag with a deadline and create a dagrun as a baseline..

Review Comment:
   ```suggestion
           # Create a new Dag with a deadline and create a dagrun as a baseline.
   ```



##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -503,9 +565,20 @@ def write_dag(
                 and existing_serialized_dag.data
                 and (existing_deadline_uuids := existing_serialized_dag.data.get("dag", {}).get("deadline"))
             ):
-                dag.data["dag"]["deadline"] = existing_deadline_uuids
-                deadline_uuid_mapping = {}
+                deadline_uuid_mapping = cls._try_reuse_deadline_uuids(
+                    existing_deadline_uuids,
+                    dag.data["dag"]["deadline"],
+                    session,
+                )
+
+                if deadline_uuid_mapping is not None:
+                    # All deadlines matched, reuse the UUIDs to preserve hash.
+                    dag.data["dag"]["deadline"] = list(deadline_uuid_mapping.keys())

Review Comment:
   if all matched, why don't we just use `existing_deadline_uuids`



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org