SameerMesiah97 commented on code in PR #72395:
URL: https://github.com/apache/airflow/pull/72395#discussion_r3908658060
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -587,19 +587,33 @@ class DagModelOperation(NamedTuple):
def find_orm_dags(self, *, session: Session) -> dict[str, DagModel]:
"""Find existing DagModel objects from DAG objects."""
+ # NOTE: These relationships are eager-loaded with selectinload (a
separate follow-up
+ # "WHERE dag_id IN (...)" query per collection) rather than joinedload
(a single query
+ # joining every collection at once). Five joinedload() calls on
one-to-many collections
+ # in a single query cause a cartesian-product row explosion: a run
against a production
+ # deployment showed 500 input dag_ids producing 3,907 result rows. The
database executed
+ # that query quickly, but the client had to receive, deserialize, and
de-duplicate all
+ # those duplicate rows (including redundant copies of DagModel's JSON
columns), which
+ # was observed causing multi-second delays. selectinload trades the
single big query for
+ # up to five smaller follow-up queries, but each returns only the rows
that actually
+ # exist, with no multiplication. See
https://github.com/apache/airflow/issues/72393.
Review Comment:
This could be more concise:
```
# Use selectinload rather than joinedload to avoid a Cartesian-product row
explosion
# when eager-loading multiple one-to-many collections. Although this issues
separate
# queries for each collection, it avoids transferring and deserializing
duplicate rows
# and JSON data. See https://github.com/apache/airflow/issues/72393.
```
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -264,7 +264,19 @@ def _statement_breakdown(counts: Counter[tuple[str, str]])
-> str:
# lapsed it rewrites it, which costs two more statements per Dag and nothing
extra per call. The
# per-call price is a file that parsed cleanly: one reporting import errors
also looks up whichever
# of them are already recorded.
-FIXED_PER_CALL = 9
+#
+# FIXED_PER_CALL includes 10 statements from
DagModelOperation.find_orm_dags(), which
+# DAG.bulk_write_to_db() calls twice per persistence call (once to look up
existing DagModels,
+# once to refetch them after flushing newly-created assets so relationships
are current). Each
+# call eager-loads five one-to-many collections (tags,
schedule_asset_references,
+# schedule_asset_alias_references, task_outlet_asset_references,
dag_owner_links) via
+# selectinload, one follow-up "WHERE dag_id IN (...)" statement per
collection, on top of the
+# base DagModel select: 6 statements x 2 calls = 12, replacing the 2
statements (1 per call) it
+# cost when all five collections were joinedload'd into one query. See
GH#72393: joinedload on
+# multiple one-to-many collections in a single query multiplies result rows (a
production case
+# saw 500 dag_ids balloon into 3,907 rows), which is far more expensive for a
client to receive
+# and deserialize than the extra round trips selectinload costs here.
Review Comment:
Same here:
```
# FIXED_PER_CALL includes 12 statements from two find_orm_dags() calls. Each
call
# executes one base query and five selectinload queries for its collections.
The extra
# queries avoid the Cartesian row explosion caused by joining multiple
one-to-many
# collections. See https://github.com/apache/airflow/issues/72393.
```
--
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]
