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 fdb8594e6bf Fix /ui/assets picking a stale "last event" under
concurrent asset writes (#71047)
fdb8594e6bf is described below
commit fdb8594e6bfa89c675ff44e86fb700cc5c77b067
Author: Takayoshi Makabe <[email protected]>
AuthorDate: Fri Aug 28 23:30:57 2026 +0900
Fix /ui/assets picking a stale "last event" under concurrent asset writes
(#71047)
* Fix /ui/assets picking a stale "last event" under concurrent asset writes
* Add Unit Test
---
.../src/airflow/api_fastapi/common/db/assets.py | 22 ++++++++---
.../api_fastapi/core_api/routes/ui/test_assets.py | 44 ++++++++++++++++++++++
2 files changed, 61 insertions(+), 5 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/db/assets.py
b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
index 8ee707fc9cd..4e7bddbd213 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
@@ -20,7 +20,7 @@ from __future__ import annotations
from collections.abc import Iterable
from typing import TYPE_CHECKING
-from sqlalchemy import func, select
+from sqlalchemy import and_, func, select
from sqlalchemy.orm import selectinload
from airflow.api_fastapi.common.db.dags import eager_load_teams
@@ -42,8 +42,20 @@ if TYPE_CHECKING:
def generate_assets_with_last_event_query() -> Select:
"""Fetch Assets outer-joined to their latest AssetEvent id/timestamp."""
- max_asset_event_id_query = (
- select(AssetEvent.asset_id,
func.max(AssetEvent.id).label("max_asset_event_id"))
+ latest_asset_event_timestamps = (
+ select(AssetEvent.asset_id,
func.max(AssetEvent.timestamp).label("last_timestamp"))
+ .group_by(AssetEvent.asset_id)
+ .subquery()
+ )
+ latest_asset_event_ids = (
+ select(AssetEvent.asset_id,
func.max(AssetEvent.id).label("last_asset_event_id"))
+ .join(
+ latest_asset_event_timestamps,
+ and_(
+ AssetEvent.asset_id ==
latest_asset_event_timestamps.c.asset_id,
+ AssetEvent.timestamp ==
latest_asset_event_timestamps.c.last_timestamp,
+ ),
+ )
.group_by(AssetEvent.asset_id)
.subquery()
)
@@ -54,8 +66,8 @@ def generate_assets_with_last_event_query() -> Select:
AssetEvent.id.label("last_asset_event_id"),
AssetEvent.timestamp.label("last_asset_event_timestamp"),
)
- .outerjoin(max_asset_event_id_query, AssetModel.id ==
max_asset_event_id_query.c.asset_id)
- .outerjoin(AssetEvent, AssetEvent.id ==
max_asset_event_id_query.c.max_asset_event_id)
+ .outerjoin(latest_asset_event_ids, AssetModel.id ==
latest_asset_event_ids.c.asset_id)
+ .outerjoin(AssetEvent, AssetEvent.id ==
latest_asset_event_ids.c.last_asset_event_id)
.options(
*eager_load_asset_reference_teams(),
selectinload(AssetModel.consuming_tasks),
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
index f91f475a18b..36b200405cc 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
@@ -569,6 +569,50 @@ class TestGetAssetsUi:
assert response.status_code == 200
assert [a["name"] for a in response.json()["assets"]] == ["newer",
"older"]
+ def test_last_asset_event_uses_latest_timestamp_not_highest_id(self,
test_client, session):
+ asset = AssetModel(name="out_of_order",
uri="s3://bucket/out_of_order", group="asset")
+ session.add(asset)
+ session.add(AssetActive.for_asset(asset))
+ session.flush()
+
+ base = pendulum.datetime(2024, 1, 1)
+ latest_event = AssetEvent(asset_id=asset.id,
timestamp=base.add(days=1))
+ highest_id_event = AssetEvent(asset_id=asset.id, timestamp=base)
+ session.add_all([latest_event, highest_id_event])
+ session.flush()
+ latest_event_id = latest_event.id
+ highest_id_event_id = highest_id_event.id
+ session.commit()
+
+ assert highest_id_event_id > latest_event_id
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ last_asset_event = response.json()["assets"][0]["last_asset_event"]
+ assert last_asset_event["id"] == latest_event_id
+ assert last_asset_event["timestamp"] == "2024-01-02T00:00:00Z"
+
+ def test_last_asset_event_breaks_timestamp_tie_by_highest_id(self,
test_client, session):
+ asset = AssetModel(name="tied_timestamp",
uri="s3://bucket/tied_timestamp", group="asset")
+ session.add(asset)
+ session.add(AssetActive.for_asset(asset))
+ session.flush()
+
+ tied_timestamp = pendulum.datetime(2024, 1, 1)
+ lower_id_event = AssetEvent(asset_id=asset.id,
timestamp=tied_timestamp)
+ higher_id_event = AssetEvent(asset_id=asset.id,
timestamp=tied_timestamp)
+ session.add_all([lower_id_event, higher_id_event])
+ session.flush()
+ higher_id_event_id = higher_id_event.id
+ session.commit()
+
+ assert higher_id_event_id > lower_id_event.id
+
+ response = test_client.get("/assets")
+ assert response.status_code == 200
+ last_asset_event = response.json()["assets"][0]["last_asset_event"]
+ assert last_asset_event["id"] == higher_id_event_id
+
def test_sort_by_group(self, test_client, session):
billing = AssetModel(name="billing_asset",
uri="s3://bucket/billing_sort", group="billing")
marketing = AssetModel(name="marketing_asset",
uri="s3://bucket/marketing_sort", group="marketing")
