[v3-1-test] fix get latest serialized_dag model query to prevent "Out of sort memory" error (#55589) (#57042)

* [v3-1-test] Fix Outlet Event Extra Data is Empty in Task Instance Success Listener (#54568) (#57031)

Co-authored-by: Kevin Yang <85313829+sjyangkevin@users.noreply.github.com>

* [v3-1-test] fix get latest serialized_dag model query to prevent "Out of sort memory" error (#55589)

* fix get latest serialized_dag model query

* fix get latest serialized_dag model query

* add db type check logic
(cherry picked from commit 757db27)

Co-authored-by: Jeongwoo Do <48639483+wjddn279@users.noreply.github.com>

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Kevin Yang <85313829+sjyangkevin@users.noreply.github.com>
Co-authored-by: Jeongwoo Do <48639483+wjddn279@users.noreply.github.com>

Author: 3 people

airflow-core/src/airflow/models/serialized_dag.py

@@ -460,6 +460,15 @@ def write_dag(
 
     @classmethod
     def latest_item_select_object(cls, dag_id):
+        from airflow.settings import engine
+
+        if engine.dialect.name == "mysql":
+            # Prevent "Out of sort memory" caused by large values in cls.data column for MySQL.
+            # Details in https://github.com/apache/airflow/pull/55589
+            latest_item_id = (
+                select(cls.id).where(cls.dag_id == dag_id).order_by(cls.created_at.desc()).limit(1)
+            )
+            return select(cls).where(cls.id == latest_item_id)
         return select(cls).where(cls.dag_id == dag_id).order_by(cls.created_at.desc()).limit(1)
 
     @classmethod