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

Code before:
            ):
                dag.data["dag"]["deadline"] = existing_deadline_uuids
                deadline_uuid_mapping = {}
            else:
                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 = False
        if dag_version:
            has_task_instances = (
                session.scalar(
                    select(literal(True)).where(exists().where(TaskInstance.dag_version_id == dag_version.id))
                )
                or False
            )

        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

            # Use direct UPDATE to avoid loading the full serialized DAG
            result = session.execute(
                update(cls)
                .where(cls.dag_version_id == dag_version.id)
                .values(
                    {
                        cls._data: new_serialized_dag._data,
                        cls._data_compressed: new_serialized_dag._data_compressed,
                        cls.dag_hash: new_serialized_dag.dag_hash,
                    }
                )
            )

            if getattr(result, "rowcount", 0) == 0:
                # No rows updated - serialized DAG doesn't exist
                return False
            # The dag_version and dag_code may not have changed, still we should
            # do the below actions:
            # Update the latest dag version

Review comment (suggestion):
```suggestion
            has_task_instances = session.scalar(
                select(exists().where(TaskInstance.dag_version_id == dag_version.id))
            )
```

Code after:
            ):
                dag.data["dag"]["deadline"] = existing_deadline_uuids
                deadline_uuid_mapping = {}
            else:
                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

            # Use direct UPDATE to avoid loading the full serialized DAG
            result = session.execute(
                update(cls)
                .where(cls.dag_version_id == dag_version.id)
                .values(
                    {
                        cls._data: new_serialized_dag._data,
                        cls._data_compressed: new_serialized_dag._data_compressed,
                        cls.dag_hash: new_serialized_dag.dag_hash,
                    }
                )
            )

            if getattr(result, "rowcount", 0) == 0:
                # No rows updated - serialized DAG doesn't exist
                return False
            # The dag_version and dag_code may not have changed, still we should
            # do the below actions:
            # Update the latest dag version
            dag_version.bundle_name = bundle_name
            dag_version.bundle_version = bundle_version
            session.merge(dag_version)