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

Code before:
                .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 latest_ser_dag:  
                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)
                return True

        dagv = DagVersion.write_dag(
            dag_id=dag.dag_id,
            bundle_name=bundle_name,
            bundle_version=bundle_version,
            session=session,
        )
        log.debug("Writing Serialized DAG: %s to the DB", dag.dag_id)
        new_serialized_dag.dag_version = dagv

Review comment (suggestion):
If there's a dag_version then we should have serdag. Can you check if dag_version.serialized_dag would work? Should work, or we can improve the DagVersion.get_latest_version to load the serdag when getting the latest version.
Something like this:
`latest_ser_dag = dag_version.serialized_dag`

Code after:
                    (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
            # 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)
            return True

        dagv = DagVersion.write_dag(