ephraimbuddy commented on code in PR #63185:
URL: https://github.com/apache/airflow/pull/63185#discussion_r3109123414
##########
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:
For 2.x upgraded DAGs, `relative_fileloc` can still be NULL because the
migration only adds it as nullable. After this bulk reassignment, the first
bundle refresh calls `deactivate_deleted_dags()`, where NULL is not in the
observed file set, so these DAGs get marked stale before they are parsed. That
defeats the “trigger immediately after migration” path and can leave DAGs
hidden if the next parse fails. The fix needs to either populate/derive
`relative_fileloc` for reassigned rows or keep legacy NULL rows out of stale
deletion until a successful parse updates them.
##########
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
Review Comment:
Reassign only to a synced bundle. `bundle_names` comes from
`_bundle_config`, but `sync_bundles_to_db()` can skip a configured bundle when
`get_bundle()` /bundle construction fails. If the first configured bundle
fails, this update can write Dag rows to a bundle that was never inserted into
`dag_bundle`, causing an FK failure and crashing Dag processor startup. The
fallback should come from bundles that were actually persisted/active after
sync.
--
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]
