You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seanmuth (via GitHub)" <gi...@apache.org> on 2026/09/01 21:17:01 UTC

[PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

seanmuth opened a new pull request, #72395:
URL: https://github.com/apache/airflow/pull/72395

   closes: #72393
   
   ## Problem
   
   `DagModelOperation.find_orm_dags` in `airflow-core/src/airflow/dag_processing/collection.py` eagerly loads five different one-to-many collections on `DagModel` (`tags`, `schedule_asset_references`, `schedule_asset_alias_references`, `task_outlet_asset_references`, `dag_owner_links`) using `joinedload()` in a single query:
   
   ```python
   stmt = with_row_locks(
       (
           select(DagModel)
           .options(joinedload(DagModel.tags, innerjoin=False))
           .where(DagModel.dag_id.in_(self.dags))
           .options(joinedload(DagModel.schedule_asset_references))
           .options(joinedload(DagModel.schedule_asset_alias_references))
           .options(joinedload(DagModel.task_outlet_asset_references))
           .options(joinedload(DagModel.dag_owner_links))
       ),
       of=DagModel,
       session=session,
   )
   ```
   
   Combining multiple `joinedload()` calls on one-to-many collections into a single query produces a cartesian product across those collections. This was confirmed live on a production deployment: 500 input `dag_id`s produced **3,907** result rows via `EXPLAIN (ANALYZE, BUFFERS)`. The database itself executed the query quickly (16ms) — the real cost lands on the client, which has to receive, deserialize, and de-duplicate (via `.unique()`) every exploded row, including redundant copies of `DagModel`'s JSON columns (`partition_mapper_info`, `asset_expression`, `deadline`) once per duplicate. This was observed causing multi-second processing delays per call in production for an async SQLAlchemy/asyncpg client, but it affects any client — sync or async — syncing a DAG set with a meaningful number of tags, asset references, or owner links.
   
   ## Fix
   
   Switch all five relationships to `selectinload()`. Instead of one big join, this issues one follow-up `WHERE dag_id IN (...)` query per collection — the same eager-loading outcome (all five collections populated on the returned `DagModel` objects), but each query returns only rows that actually exist, with no multiplication.
   
   This is a real tradeoff worth being explicit about: `find_orm_dags` is called twice per `DAG.bulk_write_to_db()` invocation (once to look up existing DagModels, once to refetch after flushing new assets), so this goes from 2 statements total (1 per call) to up to 12 (1 base + 5 selectin, times 2 calls). More round trips, but each one is cheap, returns exactly the right number of rows, and avoids the client-side deserialize/dedup cost of the exploded joined result.
   
   Also dropped `innerjoin=False` from the `tags` load — that's a `joinedload`-specific knob to prevent an inner join from silently dropping DAGs with no tags. `selectinload` has no equivalent concern since it's a separate query keyed off the dag_ids already returned by the base select, so a DAG with zero tags is unaffected either way.
   
   ## Testing
   
   - Ran the full `airflow-core/tests/unit/dag_processing/test_collection.py` and `test_manager.py` suites against both sqlite and Postgres backends — all passing.
   - `test_manager.py` has a pinned per-call SQL statement budget (`FIXED_PER_CALL`) that tracks exactly this kind of change. Updated it from 9 to 19 with an inline comment explaining the +10 (5 extra `selectin` statements per `find_orm_dags()` call, times 2 calls per persistence call), so a future statement-count regression there is easy to diagnose rather than a mystery.
   - Local verification: generated a synthetic file with 400 dynamically-generated DAGs (`globals()[dag_id] = dag` pattern), each with 5 tags, 2 task-level asset outlets, and 2 owner links, then ran it through `SerializedDAG.bulk_write_to_db` (the same path the scheduler's dag processor uses) against a local Postgres instance.
     - Confirmed the unpatched `joinedload` query returns **8,000** rows for those 400 `dag_id`s (a 20x explosion from the 5×2×2 collection sizes) — directionally consistent with, and worse than, the 500→3,907 (~7.8x) production case.
     - Confirmed the patched version persists all 400 DAGs correctly — tags, owner links, and task outlet asset references all present and correct on spot-checked DAGs — with no explosion.
     - Timed steady-state re-sync of the same 400-DAG file 5x each way on the same local Postgres instance (near-zero network latency, so this understates the real-world win): unpatched avg ~0.75s, patched avg ~0.54s (~28% faster). The gap should be substantially larger for any real deployment with actual client-DB network latency and per-row deserialization overhead, which is exactly the scenario described in the issue.
   
   🤖 Generated with [Claude Code](https://claude.com/claude-code)


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #72395:
URL: https://github.com/apache/airflow/pull/72395#issuecomment-5839358601

   Hello @seanmuth - thank you for your contributions to Apache Airflow!
   
   The Airflow community has introduced a limit of **5 open pull requests at a time** for contributors without write access to the repository. You currently have 7 open pull requests, so - as a one-time step of introducing the limit - we closed the ones where maintainers have not engaged yet:
   
   * #72395 - Use selectinload instead of joinedload in find_orm_dags to avoid row explosion
   
   These pull requests stay open because maintainers are already engaged in them - they count towards your limit:
   
   * #65602 - Add (updated_at, dag_id) index to task_instance for updated_at-range API queries
   * #69985 - Fix deadline callback data encoding in migration 0094 (scheduler CrashLoopBackOff)
   * #73142 - Task SDK: don't let exit code override a confirmed terminal state
   * #73689 - Log when a newly created Dag Run will not be scheduled due to max_active_runs
   * #73692 - Expose exceeds_max_active_runs on the Dag details API response
   * #73693 - UI: Info tooltip when a Dag's active runs exceed max_active_runs
   
   This is **not** a judgement of you or of your changes. We never told contributors before that opening many pull requests at once was a problem, so there is nothing to feel bad about - and nothing is lost: your branches, commits and the review history stay where they are.
   
   What we ask you to do is to make your **first prioritization decision**: choose which of the pull requests above matter most to you, and reopen them (up to 5 open at a time, including the ones still open) with the "Reopen pull request" button or `gh pr reopen <PR_NUMBER> --repo apache/airflow`. Reopen the ones you are ready to follow through - keep them rebased, respond to review comments and fix failing checks.
   
   While your pull requests are waiting for review, the most valuable thing you can do is help in other ways - reviewing other contributors' pull requests, helping with issues, and taking part in the discussions on the devlist and Slack.
   
   Why we introduced the limit, what it means for you and how to reopen or restore a pull request is explained in https://github.com/apache/airflow/blob/main/contributing-docs/32_open_pull_request_limit.rst.
   
   ---
   Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #72395: Use selectinload instead of joinedload in find_orm_dags to avoid row explosion
URL: https://github.com/apache/airflow/pull/72395


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on PR #72395:
URL: https://github.com/apache/airflow/pull/72395#issuecomment-5504295972

   @SameerMesiah97 addressed both — thanks for the review.
   
   **Regression test added** (`TestFindOrmDagsLoaderStrategy` in `test_collection.py`): two tests that check the actual mechanism, not just a statement count that could be blindly "corrected" back:
   - `test_find_orm_dags_uses_selectinload` — inspects the compiled statement's ORM loader options directly, fails if any of the five relationships isn't `selectin`.
   - `test_find_orm_dags_does_not_multiply_rows` — a DAG with 3 tags × 2 owner links × 2 asset refs, checked at the raw DBAPI cursor level (bypassing `.unique()`, which would otherwise hide the exact multiplication this needs to catch) — asserts exactly 1 raw row for 1 dag_id.
   
   Verified both actually catch the regression: temporarily reverted to the old multi-`joinedload()` code and confirmed both tests fail correctly (`'joined'` instead of `'selectin'`, and 12 raw rows — 3×2×2 — instead of 1), then restored the fix and confirmed green.
   
   **Comments tightened** on `collection.py` and `test_manager.py` per your suggested wording.
   
   All green on Postgres: `test_collection.py` (66 passed), `test_manager.py` (165 passed), `test_dag.py` (209 passed).


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on PR #72395:
URL: https://github.com/apache/airflow/pull/72395#issuecomment-5500836127

   test failure is unrelated: 
   ```
   Run prek install-hooks
     error: Failed to install hook `go-mod-tidy`
       caused by: Failed to install go
       caused by: Failed to download go
       caused by: Failed to download file from https://go.dev/dl/go1.25.0.linux-amd64.tar.gz
       caused by: error sending request for url (https://go.dev/dl/go1.25.0.linux-amd64.tar.gz)
       caused by: client error (Connect)
       caused by: Connection reset by peer (os error 104)
     Error: Process completed with exit code 2.
   ```
   
   can push a dummy commit to rerun or just let it stand, I'll wait for the rest of CI to finish either way.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #72395:
URL: https://github.com/apache/airflow/pull/72395#issuecomment-5500851031

   > test failure is unrelated:
   > 
   > ```
   > Run prek install-hooks
   >   error: Failed to install hook `go-mod-tidy`
   >     caused by: Failed to install go
   >     caused by: Failed to download go
   >     caused by: Failed to download file from https://go.dev/dl/go1.25.0.linux-amd64.tar.gz
   >     caused by: error sending request for url (https://go.dev/dl/go1.25.0.linux-amd64.tar.gz)
   >     caused by: client error (Connect)
   >     caused by: Connection reset by peer (os error 104)
   >   Error: Process completed with exit code 2.
   > ```
   > 
   > can push a dummy commit to rerun or just let it stand, I'll wait for the rest of CI to finish either way.
   
   I would try rebasing or just convert to draft. That usually re-triigers CI. I would recommend rebase though. 


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Use selectinload instead of joinedload in find_orm_dags to avoid row explosion [airflow]

Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on PR #72395:
URL: https://github.com/apache/airflow/pull/72395#issuecomment-5503340217

   > > test failure is unrelated:
   > > ```
   > > Run prek install-hooks
   > >   error: Failed to install hook `go-mod-tidy`
   > >     caused by: Failed to install go
   > >     caused by: Failed to download go
   > >     caused by: Failed to download file from https://go.dev/dl/go1.25.0.linux-amd64.tar.gz
   > >     caused by: error sending request for url (https://go.dev/dl/go1.25.0.linux-amd64.tar.gz)
   > >     caused by: client error (Connect)
   > >     caused by: Connection reset by peer (os error 104)
   > >   Error: Process completed with exit code 2.
   > > ```
   > > 
   > > 
   > >     
   > >       
   > >     
   > > 
   > >       
   > >     
   > > 
   > >     
   > >   
   > > can push a dummy commit to rerun or just let it stand, I'll wait for the rest of CI to finish either way.
   > 
   > I would rebase or just convert to draft. That re-triigers CI. I would recommend rebase though.
   
   tests were indeed broken (just not that one) corrected and passing now


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org