Copilot commented on code in PR #63185:
URL: https://github.com/apache/airflow/pull/63185#discussion_r3109146850
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -295,6 +296,76 @@ def _extract_and_sign_template(bundle_name: str) ->
tuple[str | None, dict]:
session.execute(delete(ParseImportError).where(ParseImportError.bundle_name ==
name))
self.log.info("Deleted import errors for bundle %s which is no
longer configured", name)
+ session.flush()
+
+ @provide_session
+ def reassign_dags_with_unconfigured_bundles(self, *, session: Session =
NEW_SESSION) -> int:
+ """
+ Reassign Dags with unconfigured bundles to the first configured bundle.
+
+ This addresses Dags that reference bundles incorrectly (i.e. the Dag's
+ ``bundle_name`` matches a value in the database, but that database
value
+ does not correspond to a user-configured bundle) as a side effect of
the
+ ``0082_3_1_0_make_bundle_name_not_nullable.py`` migration
+ (see: https://github.com/apache/airflow/issues/63323).
+
+ Instead of attempting to infer the correct bundle for each Dag during
the
+ migration, we reassign all Dags with unconfigured bundles to the first
+ configured bundle at DagFileProcessorManager startup. This relaxes the
+ "Requested bundle '{name}' is not configured." error that would
otherwise
+ occur when triggering a DagRun immediately after the migration.
+
+ This fallback is not always semantically correct in environments using
+ multiple bundles, but it is a safe, temporary measure that allows
users to
+ successfully trigger DagRuns right after the migration.
+
+ The correct Dag-to-bundle assignments will be restored by the Dag
processor
+ on the next parsing cycle.
+
+ :param session: ORM Session
+ :return: Number of Dags reassigned.
+ """
+ # Import here to avoid circular import
+ # (manager -> dag -> dagrun -> taskinstance -> dag_version -> manager)
+ from airflow.models.dag import DagModel
+
+ configured_names = self.bundle_names
+ if not configured_names:
+ self.log.info(
+ "No Dag bundles are configured; skipping reassignment of Dags
with unconfigured bundles."
+ )
+ return 0
+ default_bundle = configured_names[0]
+
+ count = cast(
+ "CursorResult",
+ session.execute(
+ update(DagModel)
+ .where(
+ or_(
+ DagModel.bundle_name.notin_(configured_names),
+ DagModel.bundle_name.is_(None),
+ )
+ )
+ .values(bundle_name=default_bundle)
+ .execution_options(synchronize_session="fetch")
Review Comment:
The bulk UPDATE uses `execution_options(synchronize_session="fetch")`, which
can issue an extra SELECT and scale poorly when many `DagModel` rows match
(e.g. large installations during upgrade). Since this method is typically
called via `@provide_session` with a fresh session at startup (and does not
rely on in-session `DagModel` state), consider using
`synchronize_session=False` (or omitting it) and, if needed for callers that
pass an existing session, explicitly expiring/reloading affected objects in
those call sites/tests.
```suggestion
.execution_options(synchronize_session=False)
```
--
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]
