ephraimbuddy commented on code in PR #63185:
URL: https://github.com/apache/airflow/pull/63185#discussion_r3160316817
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -295,6 +296,85 @@ 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()
Review Comment:
```suggestion
```
Is this necessary?
##########
airflow-core/src/airflow/models/dag.py:
##########
@@ -607,6 +607,11 @@ def deactivate_deleted_dags(
select(cls)
.where(
cls.bundle_name == bundle_name,
+ # Skip legacy rows (e.g. from 2.x upgrades) whose
relative_fileloc
+ # has not been populated yet. These DAGs have never been
parsed under
+ # the new bundle system, so they cannot be considered
"deleted". The
+ # parser will set relative_fileloc on the next successful
parse cycle.
+ cls.relative_fileloc.isnot(None),
Review Comment:
What I'm worried about here is, what if the dag was really deleted? Can it
ever be marked stale?
##########
airflow-core/src/airflow/dag_processing/bundles/manager.py:
##########
@@ -295,6 +296,85 @@ 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
+
+ # Use bundles that were actually persisted and active after
sync_bundles_to_db,
+ # not self.bundle_names from config. A configured bundle whose
construction
+ # failed is skipped by sync_bundles_to_db and must not be used as a
target
+ # (its name may not exist in the dag_bundle table, causing an FK
violation).
+ active_db_names = set(
+
session.scalars(select(DagBundleModel.name).where(DagBundleModel.active.is_(True)))
+ )
+ # Preserve config ordering so the first configured-and-active bundle
is the default.
+ active_names = [name for name in self.bundle_names if name in
active_db_names]
+ if not active_names:
+ self.log.info(
+ "No active Dag bundles in the database; skipping reassignment
of Dags "
+ "with unconfigured bundles."
+ )
+ return 0
+ default_bundle = active_names[0]
Review Comment:
This also ignores multi team boundaries and we risk moving a dag from one
team to another. WDYT? Maybe we should group by team or log clearly the movement
--
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]
