Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py (Python)

Code before:
    # This comparison is to fall back to DAG timetable when no order_by is provided
    if order_by.value == [order_by.get_primary_key_string()]:
        ordering = list(latest_dag.timetable.run_ordering)
        order_by = SortParam(
            allowed_attrs=ordering,
            model=DagRun,
        ).set_value(ordering)
    dag_runs_select_filter, _ = paginated_select(
        statement=base_query,
        order_by=order_by,
        offset=offset,
        filters=[run_after, run_type, state, triggering_user],
        limit=limit,
    )
    run_ids = list(session.scalars(dag_runs_select_filter))

    task_group_sort = get_task_group_children_getter()
    if not run_ids:
        nodes = [task_group_to_dict_grid(x) for x in task_group_sort(latest_dag.task_group)]
        return [GridNodeResponse(**n) for n in nodes]

    # Process and merge the latest serdag first
    merged_nodes: list[dict[str, Any]] = []
    nodes = [task_group_to_dict_grid(x) for x in task_group_sort(latest_dag.task_group)]
    _merge_node_dicts(merged_nodes, nodes)
    del latest_dag

    # Process serdags one by one and merge immediately to reduce memory usage.
    # Use yield_per() for streaming results and expunge each serdag after processing
    # to allow garbage collection and prevent memory buildup in the session identity map.
    serdags_query = (
        select(SerializedDagModel)
        .where(
            # Even though dag_id is filtered in base_query,
            # adding this line here can improve the performance of this endpoint
            SerializedDagModel.dag_id == dag_id,
            SerializedDagModel.id != latest_serdag_id,
            SerializedDagModel.dag_version_id.in_(
                select(TaskInstance.dag_version_id)
                .join(TaskInstance.dag_run)
                .where(
                    DagRun.id.in_(run_ids),
                )
                .distinct()
            ),
        )
        .execution_options(yield_per=5)
    )

    for serdag in session.scalars(serdags_query):
        if serdag:

Review comment (suggestion):
Explicitly deleting `latest_dag` may not effectively free memory since the DAG object could still be referenced elsewhere (e.g., within `merged_nodes` through the task_group_to_dict_grid conversion). Consider removing this line as the expunge operation on line 147 is the primary mechanism for memory management, and explicit deletion provides no guaranteed benefit here.
```suggestion

```

Code after:
    # This comparison is to fall back to DAG timetable when no order_by is provided
    if order_by.value == [order_by.get_primary_key_string()]:
        ordering = list(latest_dag.timetable.run_ordering)
        order_by = SortParam(
            allowed_attrs=ordering,
            model=DagRun,
        ).set_value(ordering)
    dag_runs_select_filter, _ = paginated_select(
        statement=base_query,
        order_by=order_by,
        offset=offset,
        filters=[run_after, run_type, state, triggering_user],
        limit=limit,
    )
    run_ids = list(session.scalars(dag_runs_select_filter))

    task_group_sort = get_task_group_children_getter()
    if not run_ids:
        nodes = [task_group_to_dict_grid(x) for x in task_group_sort(latest_dag.task_group)]
        return [GridNodeResponse(**n) for n in nodes]

    # Process and merge the latest serdag first
    merged_nodes: list[dict[str, Any]] = []
    nodes = [task_group_to_dict_grid(x) for x in task_group_sort(latest_dag.task_group)]
    _merge_node_dicts(merged_nodes, nodes)
    del latest_dag

    # Process serdags one by one and merge immediately to reduce memory usage.
    # Use yield_per() for streaming results and expunge each serdag after processing
    # to allow garbage collection and prevent memory buildup in the session identity map.
    serdags_query = (
        select(SerializedDagModel)
        .where(
            # Even though dag_id is filtered in base_query,
            # adding this line here can improve the performance of this endpoint
            SerializedDagModel.dag_id == dag_id,
            SerializedDagModel.id != latest_serdag_id,
            SerializedDagModel.dag_version_id.in_(
                select(TaskInstance.dag_version_id)
                .join(TaskInstance.dag_run)
                .where(
                    DagRun.id.in_(run_ids),
                )
                .distinct()
            ),
        )
        .execution_options(yield_per=5)  # balance between peak memory usage and round trips
    )

    for serdag in session.scalars(serdags_query):
        filtered_dag = serdag.dag