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

Code before:
        :param bundle_version: bundle version of the DAG
        :param min_update_interval: minimal interval in seconds to update serialized DAG
        :param session: ORM Session

        :returns: Boolean indicating if the DAG was written to the DB
        """
        # Checks if (Current Time - Time when the DAG was written to DB) < min_update_interval
        # If Yes, does nothing
        # If No or the DAG does not exists, updates / writes Serialized DAG to DB
        if min_update_interval is not None:
            if session.scalar(
                select(literal(True))
                .where(
                    cls.dag_id == dag.dag_id,
                    (timezone.utcnow() - timedelta(seconds=min_update_interval)) < cls.last_updated,
                )
                .select_from(cls)
            ):
                return False

        log.debug("Checking if DAG (%s) changed", dag.dag_id)
        new_serialized_dag = cls(dag)
        serialized_dag_hash = session.scalars(
            select(cls.dag_hash).where(cls.dag_id == dag.dag_id).order_by(cls.created_at.desc())
        ).first()
        dag_version = DagVersion.get_latest_version(dag.dag_id, session=session, load_serialized_dag=True)

        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

        if dag_version and not dag_version.task_instances:
            # This is for dynamic DAGs whose hashes change often. We should update
            # the serialized DAG, the dag_version, and the dag_code instead of creating
            # a new version if the dag_version has no associated task instances.

            latest_ser_dag = dag_version.serialized_dag  # Use serialized_dag directly from dag_version

            if TYPE_CHECKING:
                assert latest_ser_dag is not None

            if latest_ser_dag:
                latest_ser_dag._data = new_serialized_dag._data
                latest_ser_dag._data_compressed = new_serialized_dag._data_compressed
                latest_ser_dag.dag_hash = new_serialized_dag.dag_hash
                session.merge(latest_ser_dag)


Review comment (suggestion):
I thought about this for a while but can't help thinking the query is not enough. So, I think we should go with a direct query here like this:
```python
session.query(DagVersion).where(DagVersion.dag_id == dag.dag_id,
DagVersion.bundle_version == bundle_version, DagVersion.bundle_name==bundle_name).
options(joinedload(DagVersion.task_instances)).options(joinedload(DagVersion.serialized_dag).
order_by(DagVersion.created_at.desc()).limit(1)
```
and remove line 420 below where we do `and dag_version.bundle_name == bundle_name`

I think this is better as this is a special need that we don't do elsewhere. Apologies for not thinking it through all these while

Code after:
        :param dag: a DAG to be written into database
        :param bundle_name: bundle name of the DAG
        :param bundle_version: bundle version of the DAG
        :param min_update_interval: minimal interval in seconds to update serialized DAG
        :param session: ORM Session

        :returns: Boolean indicating if the DAG was written to the DB
        """
        # Checks if (Current Time - Time when the DAG was written to DB) < min_update_interval
        # If Yes, does nothing
        # If No or the DAG does not exists, updates / writes Serialized DAG to DB
        if min_update_interval is not None:
            if session.scalar(
                select(literal(True))
                .where(
                    cls.dag_id == dag.dag_id,
                    (timezone.utcnow() - timedelta(seconds=min_update_interval)) < cls.last_updated,
                )
                .select_from(cls)
            ):
                return False

        log.debug("Checking if DAG (%s) changed", dag.dag_id)
        new_serialized_dag = cls(dag)
        serialized_dag_hash = session.scalars(
            select(cls.dag_hash).where(cls.dag_id == dag.dag_id).order_by(cls.created_at.desc())
        ).first()
        dag_version = session.scalar(
            select(DagVersion)
            .where(DagVersion.dag_id == dag.dag_id)
            .options(joinedload(DagVersion.task_instances))
            .options(joinedload(DagVersion.serialized_dag))
            .order_by(DagVersion.created_at.desc())
            .limit(1)
        )

        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

        if dag_version and not dag_version.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
            latest_ser_dag = cls.get(dag.dag_id, session=session)
            if not latest_ser_dag:
                return False