kaxil commented on code in PR #75404:
URL: https://github.com/apache/airflow/pull/75404#discussion_r4361538948
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -4217,30 +4217,17 @@ def _update_asset_orphanage(self, *, session: Session =
NEW_SESSION) -> None:
:seealso: :meth:`AssetModelOperation.activate_assets_if_possible`.
"""
- # Group assets into orphaned=True and orphaned=False groups.
- orphaned = (
- (
- func.count(DagScheduleAssetReference.dag_id)
- + func.count(TaskOutletAssetReference.dag_id)
- + func.count(TaskInletAssetReference.dag_id)
- # An asset materialized from an ``AssetAlias`` outlet is
referenced only through
- # the alias association, not the schedule/outlet/inlet tables.
Count it so it is
- # activated (and shown in the UI) instead of being treated as
orphaned. See #58058.
- + func.count(alias_association_table.c.alias_id)
- )
- == 0
- ).label("orphaned")
- asset_reference_query = (
- select(AssetModel)
- .outerjoin(DagScheduleAssetReference)
- .outerjoin(TaskOutletAssetReference)
- .outerjoin(TaskInletAssetReference)
- .outerjoin(alias_association_table,
alias_association_table.c.asset_id == AssetModel.id)
- .group_by(AssetModel.id)
+ referenced = or_(
+ exists().where(DagScheduleAssetReference.asset_id ==
AssetModel.id),
+ exists().where(TaskOutletAssetReference.asset_id == AssetModel.id),
+ exists().where(TaskInletAssetReference.asset_id == AssetModel.id),
+ # An asset materialized from an ``AssetAlias`` outlet is
referenced only through
+ # the alias association, not the schedule/outlet/inlet tables.
Count it so it is
+ # activated (and shown in the UI) instead of being treated as
orphaned. See #58058.
+ exists().where(alias_association_table.c.asset_id ==
AssetModel.id),
)
-
- orphan_query = asset_reference_query.having(orphaned).cte()
- activate_query = asset_reference_query.having(~orphaned).cte()
+ orphan_query = select(AssetModel).where(~referenced).cte()
+ activate_query = select(AssetModel).where(referenced).cte()
Review Comment:
I measured this with the same table copies as my comment above, on Postgres
14 in the steady state (every referenced asset already in `asset_active` except
50), median of 7 runs. With 3k/6k/12k referenced assets,
`inactive_assets_query` took 316/1,295/6,150 ms as is, 283/1,097/4,163 ms with
a `(name, uri)` CTE, and 11/22/50 ms with the CTE join inside the `NOT EXISTS`
removed. All of these returned the same inactive asset ids in every run,
including one with 3,000 referenced-but-inactive assets.
So narrowing helps by 10-32%, and raising `extra` to ~20 KB per asset barely
changed that. Most of the time is the `NOT EXISTS`: it joins `anon_1` a second
time, so Postgres materializes the CTE and rescans it once per outer row (`CTE
Scan on anon_1 anon_1_1 (actual rows=2976 loops=5950)` at 6k), which makes
`inactive_assets_query` quadratic in the number of referenced assets.
The outer query already joins `asset` to the CTE on `(name, uri)`, so that
inner join adds nothing. This gives the same rows:
```python
.where(~exists().where(AssetActive.name == AssetModel.name, AssetActive.uri
== AssetModel.uri))
```
MySQL 8.0 was 35-39 ms for all four variants at 6k referenced, so this is
Postgres-specific. The inner join came in with #62114, so it could go here or
in a follow-up.
--
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]
