Repository: apache/airflow
File: airflow-core/src/airflow/models/serialized_dag.py (Python)

Code before:
        if dag.data.get("dag", {}).get("deadline"):
            # Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
            # This preserves the hash and avoids unnecessary SerializedDagModel recreations.
            existing_serialized_dag = session.scalar(
                select(cls).where(cls.dag_id == dag.dag_id).order_by(cls.created_at.desc()).limit(1)
            )

            if (
                existing_serialized_dag
                and existing_serialized_dag.data
                and (existing_deadline_uuids := existing_serialized_dag.data.get("dag", {}).get("deadline"))
            ):
                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.
                deadline_uuid_mapping = cls._generate_deadline_uuids(dag.data)
        else:
            deadline_uuid_mapping = {}

        new_serialized_dag = cls(dag)

        if (
            serialized_dag_hash == new_serialized_dag.dag_hash
            and dag_version
            and dag_version.bundle_name == bundle_name
        ):
            log.debug("Serialized DAG (%s) is unchanged. Skipping writing to DB", dag.dag_id)
            return False

        has_task_instances: bool = False
        if dag_version:
            has_task_instances = bool(
                session.scalar(select(exists().where(TaskInstance.dag_version_id == dag_version.id)))
            )

        if dag_version and not has_task_instances:
            # This is for dynamic DAGs that the hashes changes often. We should update
            # the serialized dag, the dag_version and the dag_code instead of a new version
            # if the dag_version is not associated with any task instances


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

Code after:
        if dag.data.get("dag", {}).get("deadline"):
            # Try to reuse existing deadline UUIDs if the deadline definitions haven't changed.
            # This preserves the hash and avoids unnecessary SerializedDagModel recreations.
            existing_serialized_dag = session.scalar(
                select(cls).where(cls.dag_id == dag.dag_id).order_by(cls.created_at.desc()).limit(1)
            )

            if (
                existing_serialized_dag
                and existing_serialized_dag.data
                and (existing_deadline_uuids := existing_serialized_dag.data.get("dag", {}).get("deadline"))
            ):
                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.
                    # Clear the mapping since the alert rows already exist in the DB;
                    # no need to delete and recreate identical records.
                    dag.data["dag"]["deadline"] = existing_deadline_uuids
                    deadline_uuid_mapping = {}
                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.
                deadline_uuid_mapping = cls._generate_deadline_uuids(dag.data)
        else:
            deadline_uuid_mapping = {}

        new_serialized_dag = cls(dag)

        if (
            serialized_dag_hash == new_serialized_dag.dag_hash
            and dag_version
            and dag_version.bundle_name == bundle_name
        ):
            log.debug("Serialized DAG (%s) is unchanged. Skipping writing to DB", dag.dag_id)
            return False

        has_task_instances: bool = False
        if dag_version:
            has_task_instances = bool(
                session.scalar(select(exists().where(TaskInstance.dag_version_id == dag_version.id)))
            )

        if dag_version and not has_task_instances:
            # This is for dynamic DAGs that the hashes changes often. We should update