steveahnahn commented on code in PR #68595:
URL: https://github.com/apache/airflow/pull/68595#discussion_r3425021877
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3690,16 +3694,24 @@ def _cleanup_orphaned_asset_state_store(*, session:
Session) -> None:
"""
Delete asset_state_store rows for assets no longer active in any Dag.
- When _orphan_unreferenced_assets removes an asset from asset_active,
its
- asset_state_store rows become unreachable — no task can write to them
anymore.
- This runs in the same pass as asset orphanage to keep the table clean.
+ Bounded to ORPHANED_ASSET_STATE_STORE_CLEANUP_BATCH_SIZE orphaned
assets per
+ tick so a large backlog cannot become one unbounded delete in the
scheduler
+ loop; the remainder drains on later orphanage ticks.
"""
active_asset_ids = select(AssetModel.id).join(
AssetActive,
(AssetActive.name == AssetModel.name) & (AssetActive.uri ==
AssetModel.uri),
)
+ orphaned_asset_ids = session.scalars(
+ select(AssetStateStoreModel.asset_id)
+ .where(AssetStateStoreModel.asset_id.not_in(active_asset_ids))
+ .distinct()
Review Comment:
It's bounded either way. The select has a .limit(BATCH_SIZE), so at most
BATCH_SIZE rows are ever pulled in and the dedup is over that capped set, not
the full table, so there's no full scan/sort to worry about.
--
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]
