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 a3b1e13827f Eager-load Assets list relationships with selectinload and
refresh it after new events (#70997)
a3b1e13827f is described below
commit a3b1e13827f18e3012568818e33149d45b1a4057
Author: Brent Bovenzi <[email protected]>
AuthorDate: Tue Aug 11 05:04:05 2026 -0400
Eager-load Assets list relationships with selectinload and refresh it after
new events (#70997)
The Assets list eager-loads five relationships. subqueryload re-runs the
whole
last-event query — including its max(asset_event.id) GROUP BY aggregation —
as an
inner subquery once per relationship, so a single list page issued seven
aggregations of asset_event. selectinload emits a plain WHERE asset_id IN
(page
ids) instead, matching the Dags list helper.
The query-count guard is also strengthened to give assets a watcher (with a
trigger) and an alias, so it actually covers the relationships and the
nested
watcher -> trigger join it exists to protect.
Creating an asset event now invalidates the Assets list cache, so the row's
last
event timestamp — and its position under the default last-event sort —
update
instead of going stale.
---
.../src/airflow/api_fastapi/common/db/assets.py | 12 +++++------
.../src/pages/Asset/CreateAssetEventModal.test.tsx | 17 ++++++++++++++-
.../ui/src/pages/Asset/CreateAssetEventModal.tsx | 8 ++++++-
.../api_fastapi/core_api/routes/ui/test_assets.py | 25 ++++++++++++++++------
4 files changed, 48 insertions(+), 14 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 e393ea19060..b0ec961d864 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 typing import TYPE_CHECKING
from sqlalchemy import func, select
-from sqlalchemy.orm import subqueryload
+from sqlalchemy.orm import selectinload
from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel
@@ -45,10 +45,10 @@ def generate_assets_with_last_event_query() -> Select:
.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)
.options(
- subqueryload(AssetModel.scheduled_dags),
- subqueryload(AssetModel.producing_tasks),
- subqueryload(AssetModel.consuming_tasks),
- subqueryload(AssetModel.aliases),
-
subqueryload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
+ selectinload(AssetModel.scheduled_dags),
+ selectinload(AssetModel.producing_tasks),
+ selectinload(AssetModel.consuming_tasks),
+ selectinload(AssetModel.aliases),
+
selectinload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
)
)
diff --git
a/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.test.tsx
b/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.test.tsx
index a7e91e64c05..9784f59ed88 100644
--- a/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.test.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.test.tsx
@@ -16,13 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
+import { QueryClient } from "@tanstack/react-query";
import "@testing-library/jest-dom";
import { fireEvent, render, screen } from "@testing-library/react";
import type { ReactNode } from "react";
import { beforeEach, describe, expect, it, vi } from "vitest";
import type * as OpenapiQueries from "openapi/queries";
-import type { AssetResponse, DAGDetailsResponse } from
"openapi/requests/types.gen";
+import type { AssetEventResponse, AssetResponse, DAGDetailsResponse } from
"openapi/requests/types.gen";
import type { DagRunTriggerParams } from "src/components/TriggerDag/types";
import type * as Ui from "src/components/ui";
import { Wrapper } from "src/utils/Wrapper";
@@ -90,6 +91,7 @@ vi.mock("openapi/queries", async (importOriginal) => {
const {
useAssetServiceCreateAssetEvent,
+ useAssetServiceGetAssetsUiKey,
useAssetServiceMaterializeAsset,
useDagServiceGetDagDetails,
useDependenciesServiceGetDependencies,
@@ -169,6 +171,19 @@ describe("CreateAssetEventModal", () => {
});
});
+ it("invalidates the assets list cache after creating an event", async () => {
+ const invalidateSpy = vi.spyOn(QueryClient.prototype,
"invalidateQueries").mockResolvedValue();
+
+ render(<CreateAssetEventModal asset={asset} onClose={vi.fn()} open />, {
wrapper: Wrapper });
+
+ const onSuccess =
vi.mocked(useAssetServiceCreateAssetEvent).mock.calls.at(-1)?.[0]?.onSuccess as
+ ((data: AssetEventResponse) => Promise<void>) | undefined;
+
+ await onSuccess?.({ id: 1 } as unknown as AssetEventResponse);
+
+ expect(invalidateSpy).toHaveBeenCalledWith({ queryKey:
[useAssetServiceGetAssetsUiKey] });
+ });
+
it("sends the entered manual partition key as-is", () => {
render(<CreateAssetEventModal asset={asset} onClose={vi.fn()} open />, {
wrapper: Wrapper });
diff --git
a/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.tsx
b/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.tsx
index b4e64b0f1d8..76a291d4667 100644
--- a/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Asset/CreateAssetEventModal.tsx
@@ -25,6 +25,7 @@ import { FiPlay } from "react-icons/fi";
import {
useAssetServiceCreateAssetEvent,
UseAssetServiceGetAssetEventsKeyFn,
+ useAssetServiceGetAssetsUiKey,
useAssetServiceMaterializeAsset,
UseDagRunServiceGetDagRunsKeyFn,
useDagServiceGetDagDetails,
@@ -78,7 +79,12 @@ export const CreateAssetEventModal = ({ asset, onClose, open
}: Props) => {
setPartitionKey(undefined);
onClose();
- let queryKeys = [UseAssetServiceGetAssetEventsKeyFn({ assetId: asset.id },
[{ assetId: asset.id }])];
+ let queryKeys = [
+ UseAssetServiceGetAssetEventsKeyFn({ assetId: asset.id }, [{ assetId:
asset.id }]),
+ // The Assets list defaults to sorting by last asset event, so a new
event changes both the
+ // displayed timestamp and the row's position; refresh it for manual
events and materializes.
+ [useAssetServiceGetAssetsUiKey],
+ ];
if ("dag_run_id" in response) {
const dagId = response.dag_id;
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 2427729035f..9cee5f1ac8c 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
@@ -31,9 +31,11 @@ from airflow.models.asset import (
AssetEvent,
AssetModel,
AssetPartitionDagRun,
+ AssetWatcherModel,
DagScheduleAssetReference,
PartitionedAssetKeyLog,
)
+from airflow.models.trigger import Trigger
from airflow.partition_mappers.base import RollupMapper
from airflow.partition_mappers.temporal import StartOfHourMapper
from airflow.partition_mappers.window import HourWindow
@@ -756,12 +758,23 @@ class TestGetAssetsUi:
def test_query_count(self, test_client, session):
"""The asset relationships are eager-loaded, so the query count stays
fixed regardless of
- how many assets are returned (a lazy-loading regression would issue
queries per asset)."""
- for i in range(5):
- asset = AssetModel(name=f"asset{i}", uri=f"s3://bucket/asset{i}",
group="asset")
- session.add(asset)
- session.add(AssetActive.for_asset(asset))
+ how many assets are returned (a lazy-loading regression would issue
queries per asset).
+
+ At least two assets carry a watcher (with a trigger) and an alias so
the count guards the
+ ``selectinload`` of every relationship, including the nested
``watchers -> trigger`` join.
+ """
+ assets = [AssetModel(name=f"asset{i}", uri=f"s3://bucket/asset{i}",
group="asset") for i in range(5)]
+ session.add_all(assets)
+ session.add_all(AssetActive.for_asset(asset) for asset in assets)
+ session.flush()
+
+ for i in range(2):
+ trigger =
Trigger(classpath=f"airflow.triggers.testing.TestTrigger{i}", kwargs={})
+ session.add(trigger)
+ session.flush()
+ assets[i].watchers.append(AssetWatcherModel(name=f"watcher{i}",
trigger_id=trigger.id))
+ assets[i].aliases.append(AssetAliasModel(name=f"alias{i}",
group=""))
session.commit()
- with assert_queries_count(7):
+ with assert_queries_count(8):
assert test_client.get("/assets").status_code == 200
