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

Code before:
        :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)

        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 TYPE_CHECKING:
                assert latest_ser_dag is not None
            # Update the serialized DAG with the new_serialized_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)
            # 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)
            # Update the latest DagCode
            DagCode.update_source_code(dag_id=dag.dag_id, fileloc=dag.fileloc, session=session)

Review comment (bug):
This is wrong -- why are we loading the entire serialized dag again before checking if the hash has changed or not.

cc @ephraimbuddy 

Code after:
        :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
            # Update the serialized DAG with the new_serialized_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)
            # The dag_version and dag_code may not have changed, still we should