shahar1 commented on code in PR #74915:
URL: https://github.com/apache/airflow/pull/74915#discussion_r4506549764
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -464,11 +607,12 @@ def _extract_and_sign_template(bundle_name: str) ->
tuple[str | None, dict]:
from airflow.models.errors import ParseImportError
for name, bundle in stored.items():
- bundle.active = False
- bundle.teams = []
- self.log.warning("DAG bundle %s is no longer found in config and
has been disabled", name)
+ if bundle.active is not False:
+ bundle.active = False
+ bundle.teams = []
+ self.log.warning("DAG bundle %s is no longer found in config
and has been disabled", name)
+
session.execute(delete(ParseImportError).where(ParseImportError.bundle_name ==
name))
Review Comment:
**minor** — With reload enabled, `_reconcile_bundles()` calls
`sync_bundles_to_db()` on every periodic check
(`bundle_refresh_check_interval`, default 5 s). This loop now issues one
`DELETE FROM import_error WHERE bundle_name = ?` for **every** stored bundle
that is not configured, not only the ones deactivated in this pass.
`DagBundleModel` rows are never removed, so the list grows with every bundle
ever removed, including bundles that were removed before reload was turned on.
`import_error.bundle_name` has no index (`airflow/models/errors.py`). The
unfiltered Dag processor therefore runs N full scans of `import_error` every 5
s in its main loop, inside one transaction. Before this PR the same delete ran
once, at startup.
`AGENTS.md`:
> Bulk `DELETE`/`UPDATE` in the scheduler loop or any synchronous interval
task (e.g. `call_regular_interval` callbacks) must be batched with `LIMIT` and
committed between batches — never issue a single unbounded bulk write against a
user-driven table. Unbounded bulk writes hold row locks for the entire
transaction (blocking concurrent writers) and stall the scheduler main loop.
Filter columns used by the cleanup must be indexed.
**Suggestion:** Limit the cleanup to bundles that this pass deactivates, or
that were deactivated recently (that covers late parse results). Use one
`bundle_name.in_(...)` statement rather than one per bundle. Another option is
to drop late import errors for inactive bundles where the parse result is
written. The test in `test_sync_bundles_to_db` ("Another processor may publish
an error after deactivation") would need to follow whichever option you choose.
--
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]
