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

Code before:
            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.scalars(
            select(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())
        ).first()

        if serialized_dag_hash == new_serialized_dag.dag_hash and dag_version:
            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)

                # Update the dag_version details
                dag_version.bundle_name = bundle_name
                dag_version.bundle_version = bundle_version
                session.merge(dag_version)

Review comment (suggestion):
```suggestion
        ).limit(1)
```
limit is actually faster

Code after:
        # 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
            # do the below actions:
            # Update the latest dag version
            dag_version.bundle_name = bundle_name
            dag_version.bundle_version = bundle_version