This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 32f8527be1b Activate assets materialized from an AssetAlias so they
appear in the Assets tab (#71555)
32f8527be1b is described below
commit 32f8527be1bf032b607e78de5a446d7135f2f2b4
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Thu Aug 20 16:03:45 2026 +0200
Activate assets materialized from an AssetAlias so they appear in the
Assets tab (#71555)
An asset created at runtime by attaching it to an AssetAlias outlet via
Metadata is
referenced only through the alias association, not the
schedule/outlet/inlet tables,
so the asset-orphanage pass treated it as orphaned and never activated it.
The Assets
tab defaults to only_active, so the asset (and its alias) never showed up
despite
having events. Count the alias association as a reference so such assets
are activated
like any other.
Fixes #58058.
---
.../src/airflow/jobs/scheduler_job_runner.py | 10 ++++++--
airflow-core/tests/unit/jobs/test_scheduler_job.py | 29 ++++++++++++++++++++++
2 files changed, 37 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 2153f426758..d744de41816 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -85,6 +85,7 @@ from airflow.models.asset import (
PartitionedAssetKeyLog,
TaskInletAssetReference,
TaskOutletAssetReference,
+ alias_association_table,
association_table,
)
from airflow.models.asset_state_store import AssetStateStoreModel
@@ -3843,8 +3844,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
Check assets orphanization and update their active entry.
An orphaned asset is no longer referenced in any DAG schedule
parameters,
- task outlets, or task inlets. Active assets (non-orphaned) have
entries in
- AssetActive and must have unique names and URIs.
+ task outlets, task inlets, or alias associations. Active assets
(non-orphaned)
+ have entries in AssetActive and must have unique names and URIs.
:seealso: :meth:`AssetModelOperation.activate_assets_if_possible`.
"""
@@ -3854,6 +3855,10 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
func.count(DagScheduleAssetReference.dag_id)
+ func.count(TaskOutletAssetReference.dag_id)
+ func.count(TaskInletAssetReference.dag_id)
+ # An asset materialized from an ``AssetAlias`` outlet is
referenced only through
+ # the alias association, not the schedule/outlet/inlet tables.
Count it so it is
+ # activated (and shown in the UI) instead of being treated as
orphaned. See #58058.
+ + func.count(alias_association_table.c.alias_id)
)
== 0
).label("orphaned")
@@ -3862,6 +3867,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.outerjoin(DagScheduleAssetReference)
.outerjoin(TaskOutletAssetReference)
.outerjoin(TaskInletAssetReference)
+ .outerjoin(alias_association_table,
alias_association_table.c.asset_id == AssetModel.id)
.group_by(AssetModel.id)
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index fd3b0fd0964..88fcb0555ce 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -8844,6 +8844,35 @@ class TestSchedulerJob:
assert active == [asset1, asset3, asset5]
assert orphaned == [asset2, asset4]
+ def test_asset_orphaning_keeps_alias_materialized_asset_active(self,
session):
+ """An asset linked only through an ``AssetAlias`` association must
stay active, not orphaned.
+
+ Reproduces #58058: a task with an ``AssetAlias`` outlet materializes a
concrete asset at
+ runtime (via ``Metadata``). It has no schedule/outlet/inlet reference
— only the alias
+ association — so the orphanage pass used to treat it as orphaned,
leaving it out of the
+ Assets tab despite having events and a live alias.
+ """
+ self.job_runner = SchedulerJobRunner(job=Job())
+
+ asset = AssetModel(uri="test://alias_materialized",
name="alias_materialized_asset", group="asset")
+ alias = AssetAliasModel(name="materializing_alias", group="asset")
+ asset.aliases.append(alias)
+ session.add_all([asset, alias])
+ session.flush()
+
+ # Referenced only via the alias association, so inactive before the
pass.
+ orphaned, active = self._find_assets_activation(session)
+ assert asset in orphaned
+ assert asset not in active
+
+ self.job_runner._update_asset_orphanage(session=session)
+ session.flush()
+
+ # The alias association now counts as a reference, so the asset is
activated.
+ orphaned, active = self._find_assets_activation(session)
+ assert asset in active
+ assert asset not in orphaned
+
def test_asset_orphaning_ignore_orphaned_assets(self, dag_maker, session):
self.job_runner = SchedulerJobRunner(job=Job())
