seanmuth commented on code in PR #69832:
URL: https://github.com/apache/airflow/pull/69832#discussion_r3751413926
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py:
##########
@@ -193,11 +194,23 @@ def get_dag_structure(
_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)
+ # Identify the historical serialized Dags for the visible runs (ids only,
cheap),
+ # then process them in small batches so we never hold a DB connection open
across
+ # the CPU-bound deserialization/merge below.
+ #
+ # Deserializing (``serdag.dag``) and merging a large DAG's task groups is
+ # expensive. Streaming the rows with a server-side cursor (``yield_per``)
and
+ # deserializing between fetches keeps the transaction — and, under
PgBouncer
+ # transaction pooling, the pooled server connection — pinned for the entire
+ # render; at scale that exhausts the pool and starves task-instance
heartbeats.
+ # See https://github.com/apache/airflow/issues/65712.
+ #
+ # Each batch is loaded and detached in its own short-lived session that is
closed
+ # before the batch is deserialized, so the connection is released during
the CPU
+ # work while peak memory stays bounded to one batch
(``SerializedDagModel.dag``
+ # only reads the already-loaded ``data`` column, so it works on detached
rows).
Review Comment:
Trimmed to exactly your suggested wording in 4aef9fa789. Thanks.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py:
##########
@@ -212,24 +225,33 @@ def get_dag_structure(
.distinct()
),
)
- .execution_options(yield_per=5) # balance between peak memory usage
and round trips
+ .order_by(SerializedDagModel.id)
Review Comment:
Good catch, dropped it in 4aef9fa789. It wasn't necessary — the original
code had no explicit ordering either, and `_merge_node_dicts` is
order-sensitive for nodes that differ across historical Dag versions (first
write wins for structure), so adding an ordering guarantee here would have been
an unreviewed behavior change relative to what shipped before. This restores
the exact prior (unordered) behavior.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py:
##########
@@ -212,24 +225,33 @@ def get_dag_structure(
.distinct()
),
)
- .execution_options(yield_per=5) # balance between peak memory usage
and round trips
+ .order_by(SerializedDagModel.id)
)
-
- 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
+ serdag_ids = list(session.scalars(serdag_id_query))
+ # Release the request session's transaction/connection before the batched
work.
+ session.commit()
Review Comment:
Agreed, swapped to `session.close()` in 4aef9fa789 — this session does no
writes here, so close() is more accurate and releases the connection the same
way. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
