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

Code before:
    # 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:
            filtered_dag = serdag.dag
            # Apply the same filtering to historical DAG versions
            if root:
                filtered_dag = filtered_dag.partial_subset(
                    task_ids=root,
                    include_upstream=include_upstream,
                    include_downstream=include_downstream,
                    depth=depth,
                )
            # Merge immediately instead of collecting all DAGs in memory
            nodes = [task_group_to_dict_grid(x) for x in task_group_sort(filtered_dag.task_group)]
            _merge_node_dicts(merged_nodes, nodes)

            session.expunge(serdag)  # to allow garbage collection

    return [GridNodeResponse(**n) for n in merged_nodes]


@grid_router.get(
    "/runs/{dag_id}",
    responses=create_openapi_http_exception_doc(

Review comment (suggestion):
This seems like a reasonable balance between not loading too much in memory, but also not doing too many round trips.

Code after:
    # 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
        # Apply the same filtering to historical DAG versions
        if root:
            filtered_dag = filtered_dag.partial_subset(
                task_ids=root,
                include_upstream=include_upstream,
                include_downstream=include_downstream,
                depth=depth,
            )
        # Merge immediately instead of collecting all DAGs in memory
        nodes = [task_group_to_dict_grid(x) for x in task_group_sort(filtered_dag.task_group)]
        _merge_node_dicts(merged_nodes, nodes)

        session.expunge(serdag)  # to allow garbage collection

    return [GridNodeResponse(**n) for n in merged_nodes]


@grid_router.get(
    "/runs/{dag_id}",
    responses=create_openapi_http_exception_doc(
        [