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 3f66d0f2da5 Mark only a run's most recent asset event as triggering it
(#71441)
3f66d0f2da5 is described below
commit 3f66d0f2da5c81e010ea280b24f47175cfef02f4
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Thu Aug 13 12:46:18 2026 +0200
Mark only a run's most recent asset event as triggering it (#71441)
* Mark only a run's most recent asset event as triggering it
An asset event is linked to every dag run that consumed it, so a run that
consumed several events showed "Triggered Dag Run" on each of them — in both
the Asset Events view and a dag run's upstream asset events view — making it
look as though more runs had been created than actually were.
Only a run's most recent consumed event triggers it; the earlier ones are
merely included in the run. Surfacing that distinction makes the attribution
match what actually happened.
closes: #56749
* Match asset event triggering by id and add an N+1 query-count test
Address review feedback: match each created-dagrun reference to its source
event by id instead of relying on the positional order of two parallel
lists.
The same asset event can trigger one dag run while being merely included in
another (an asset can fan out to several consumers), so the flag is resolved
per (event, run) pair. The endpoints always populate it, so it is a required
field with no default. Also add a query-count assertion so both asset-event
endpoints are guarded against an N+1 as the number of consumed events grows.
* Small adjustments
---
.../src/airflow/api_fastapi/common/db/assets.py | 44 ++++++++++-
.../api_fastapi/core_api/datamodels/assets.py | 7 ++
.../core_api/openapi/v2-rest-api-generated.yaml | 7 ++
.../api_fastapi/core_api/routes/public/assets.py | 3 +-
.../api_fastapi/core_api/routes/public/dag_run.py | 8 +-
.../api_fastapi/core_api/services/public/assets.py | 45 +++++++++++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 7 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 4 +
.../airflow/ui/public/i18n/locales/en/common.json | 1 +
.../src/components/Assets/TriggeredRuns.test.tsx | 86 ++++++++++++++++++++++
.../ui/src/components/Assets/TriggeredRuns.tsx | 42 +++++++++--
.../core_api/routes/public/test_assets.py | 34 ++++++++-
.../core_api/routes/public/test_dag_run.py | 55 +++++++++++++-
.../src/airflowctl/api/datamodels/generated.py | 7 ++
.../tests/airflow_ctl/api/test_operations.py | 1 +
15 files changed, 336 insertions(+), 15 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 b0ec961d864..3e54c8dcabb 100644
--- a/airflow-core/src/airflow/api_fastapi/common/db/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
@@ -17,14 +17,17 @@
from __future__ import annotations
+from collections.abc import Iterable
from typing import TYPE_CHECKING
from sqlalchemy import func, select
from sqlalchemy.orm import selectinload
-from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel
+from airflow.models.asset import AssetEvent, AssetModel, AssetWatcherModel,
association_table
+from airflow.models.dagrun import DagRun
if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
from sqlalchemy.sql import Select
@@ -52,3 +55,42 @@ def generate_assets_with_last_event_query() -> Select:
selectinload(AssetModel.watchers).joinedload(AssetWatcherModel.trigger),
)
)
+
+
+def resolve_triggering_event_refs(
+ events: Iterable[AssetEvent], *, session: Session
+) -> set[tuple[int, str, str]]:
+ """
+ Return ``(event_id, dag_id, run_id)`` tuples where the asset event
actually triggered the run.
+
+ A dag run consumes every asset event queued when it is created, but only
the run's most recent
+ consumed event triggers it; earlier consumed events are merely included.
+ """
+ run_ids = {run.id for event in events for run in event.created_dagruns}
+ if not run_ids:
+ return set()
+ ranked_events = (
+ select(
+ association_table.c.event_id,
+ DagRun.dag_id,
+ DagRun.run_id,
+ func.row_number()
+ .over(
+ partition_by=association_table.c.dag_run_id,
+ order_by=(AssetEvent.timestamp.desc(), AssetEvent.id.desc()),
+ )
+ .label("rank"),
+ )
+ .join(AssetEvent, AssetEvent.id == association_table.c.event_id)
+ .join(DagRun, DagRun.id == association_table.c.dag_run_id)
+ .where(association_table.c.dag_run_id.in_(run_ids))
+ .subquery()
+ )
+ return {
+ (event_id, dag_id, run_id)
+ for event_id, dag_id, run_id in session.execute(
+ select(ranked_events.c.event_id, ranked_events.c.dag_id,
ranked_events.c.run_id).where(
+ ranked_events.c.rank == 1
+ )
+ )
+ }
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
index 8966891cf8b..fea14b36755 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py
@@ -170,6 +170,13 @@ class DagRunAssetReference(StrictBaseModel):
data_interval_start: datetime | None
data_interval_end: datetime | None
partition_key: str | None
+ triggering: bool = Field(
+ description=(
+ "Whether this asset event triggered the referenced dag run. Only a
run's most recent "
+ "consumed asset event triggers it; earlier consumed events are
included in the run but "
+ "did not trigger it."
+ ),
+ )
class AssetEventResponse(BaseModel):
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 5200635ec9e..b3edb5986f4 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -13933,6 +13933,12 @@ components:
- type: string
- type: 'null'
title: Partition Key
+ triggering:
+ type: boolean
+ title: Triggering
+ description: Whether this asset event triggered the referenced dag
run.
+ Only a run's most recent consumed asset event triggers it; earlier
consumed
+ events are included in the run but did not trigger it.
additionalProperties: false
type: object
required:
@@ -13945,6 +13951,7 @@ components:
- data_interval_start
- data_interval_end
- partition_key
+ - triggering
title: DagRunAssetReference
description: DagRun serializer for asset responses.
DagRunMutableStates:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
index 830d4edaeed..d4363b37bc3 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -74,6 +74,7 @@ from airflow.api_fastapi.core_api.security import (
requires_access_asset_alias,
requires_access_dag,
)
+from airflow.api_fastapi.core_api.services.public.assets import
serialize_asset_events
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.assets.manager import asset_manager
from airflow.configuration import conf
@@ -378,7 +379,7 @@ def get_asset_events(
assets_events = session.scalars(assets_event_select).all()
return AssetEventCollectionResponse(
- asset_events=assets_events,
+ asset_events=serialize_asset_events(assets_events, session=session),
total_entries=total_entries,
)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
index d43e3b56fd8..b0b2b922c16 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py
@@ -106,6 +106,7 @@ from airflow.api_fastapi.core_api.security import (
requires_access_dag_run_bulk,
requires_access_dag_run_clear_bulk,
)
+from airflow.api_fastapi.core_api.services.public.assets import
serialize_asset_events
from airflow.api_fastapi.core_api.services.public.dag_run import (
BulkDagRunService,
DagRunWaiter,
@@ -287,7 +288,10 @@ def get_upstream_asset_events(
DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id,
)
-
.options(joinedload(DagRun.consumed_asset_events).joinedload(AssetEvent.asset))
+ .options(
+
joinedload(DagRun.consumed_asset_events).joinedload(AssetEvent.asset),
+
joinedload(DagRun.consumed_asset_events).subqueryload(AssetEvent.created_dagruns),
+ )
)
if dag_run is None:
raise HTTPException(
@@ -296,7 +300,7 @@ def get_upstream_asset_events(
)
events = dag_run.consumed_asset_events
return AssetEventCollectionResponse(
- asset_events=events,
+ asset_events=serialize_asset_events(events, session=session),
total_entries=len(events),
)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
new file mode 100644
index 00000000000..f110f15e1fd
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+from airflow.api_fastapi.common.db.assets import resolve_triggering_event_refs
+from airflow.api_fastapi.core_api.datamodels.assets import AssetEventResponse
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+ from airflow.models.asset import AssetEvent
+
+
+def serialize_asset_events(events: Sequence[AssetEvent], *, session: Session)
-> list[AssetEventResponse]:
+ """
+ Serialize asset events, flagging on each created dag run whether this
event triggered it.
+
+ Only a run's most recent consumed event triggers it; the rest were merely
included. The flag is
+ resolved per ``(event_id, dag_id, run_id)`` tuple and set on the run
before validation so the response
+ carries it.
+ """
+ triggering_refs = resolve_triggering_event_refs(events, session=session)
+ asset_events = []
+ for event in events:
+ for run in event.created_dagruns:
+ run.triggering = (event.id, run.dag_id, run.run_id) in
triggering_refs
+ asset_events.append(AssetEventResponse.model_validate(event))
+ return asset_events
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index d9afa239aa2..1388b6f37f4 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -4496,11 +4496,16 @@ export const $DagRunAssetReference = {
}
],
title: 'Partition Key'
+ },
+ triggering: {
+ type: 'boolean',
+ title: 'Triggering',
+ description: "Whether this asset event triggered the referenced
dag run. Only a run's most recent consumed asset event triggers it; earlier
consumed events are included in the run but did not trigger it."
}
},
additionalProperties: false,
type: 'object',
- required: ['run_id', 'dag_id', 'logical_date', 'start_date', 'end_date',
'state', 'data_interval_start', 'data_interval_end', 'partition_key'],
+ required: ['run_id', 'dag_id', 'logical_date', 'start_date', 'end_date',
'state', 'data_interval_start', 'data_interval_end', 'partition_key',
'triggering'],
title: 'DagRunAssetReference',
description: 'DagRun serializer for asset responses.'
} as const;
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index cf8c86e997c..8136b7c0d56 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1156,6 +1156,10 @@ export type DagRunAssetReference = {
data_interval_start: string | null;
data_interval_end: string | null;
partition_key: string | null;
+ /**
+ * Whether this asset event triggered the referenced dag run. Only a run's
most recent consumed asset event triggers it; earlier consumed events are
included in the run but did not trigger it.
+ */
+ triggering: boolean;
};
/**
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
index dc3ab657d49..684a0a674de 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json
@@ -151,6 +151,7 @@
"tooltip": "Press {{hotkey}} for fullscreen"
},
"generateToken": "Generate Token",
+ "includedIn": "Included in",
"key": "Key",
"logicalDate": "Logical Date",
"logout": "Logout",
diff --git
a/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.test.tsx
b/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.test.tsx
new file mode 100644
index 00000000000..b6e79146011
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.test.tsx
@@ -0,0 +1,86 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import "@testing-library/jest-dom";
+import { render, screen } from "@testing-library/react";
+import { describe, expect, it } from "vitest";
+
+import type { DagRunAssetReference } from "openapi/requests/types.gen";
+import { Wrapper } from "src/utils/Wrapper";
+
+import { TriggeredRuns } from "./TriggeredRuns";
+
+const makeRun = (overrides: Partial<DagRunAssetReference>):
DagRunAssetReference =>
+ ({
+ dag_id: "dag_1",
+ data_interval_end: null,
+ data_interval_start: null,
+ end_date: null,
+ logical_date: null,
+ partition_key: null,
+ run_id: "run_1",
+ start_date: "2025-01-01T00:00:00Z",
+ state: "success",
+ triggering: true,
+ ...overrides,
+ }) satisfies DagRunAssetReference;
+
+describe("TriggeredRuns", () => {
+ it("labels a triggering run as triggered", () => {
+ render(<TriggeredRuns dagRuns={[makeRun({ triggering: true })]} />, {
wrapper: Wrapper });
+
+ expect(screen.getByText(/triggered dagRun_one/u)).toBeInTheDocument();
+ expect(screen.queryByText(/includedIn/u)).not.toBeInTheDocument();
+ });
+
+ it("labels a non-triggering run as included", () => {
+ render(<TriggeredRuns dagRuns={[makeRun({ triggering: false })]} />, {
wrapper: Wrapper });
+
+ expect(screen.getByText(/includedIn dagRun_one/u)).toBeInTheDocument();
+ expect(screen.queryByText(/triggered dagRun/u)).not.toBeInTheDocument();
+ });
+
+ it("splits a mix of triggering and included runs into separate labels", ()
=> {
+ render(
+ <TriggeredRuns
+ dagRuns={[
+ makeRun({ dag_id: "dag_triggered", run_id: "r1", triggering: true }),
+ makeRun({ dag_id: "dag_included", run_id: "r2", triggering: false }),
+ ]}
+ />,
+ { wrapper: Wrapper },
+ );
+
+ expect(screen.getByText(/triggered dagRun_one/u)).toBeInTheDocument();
+ expect(screen.getByText(/includedIn dagRun_one/u)).toBeInTheDocument();
+ });
+
+ it("groups multiple included runs behind a single count label", () => {
+ render(
+ <TriggeredRuns
+ dagRuns={[
+ makeRun({ dag_id: "dag_a", run_id: "r1", triggering: false }),
+ makeRun({ dag_id: "dag_b", run_id: "r2", triggering: false }),
+ ]}
+ />,
+ { wrapper: Wrapper },
+ );
+
+ expect(screen.getByText(/2 includedIn dagRun_other/u)).toBeInTheDocument();
+ });
+});
diff --git
a/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.tsx
b/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.tsx
index c5aab95ad57..6f7499791f2 100644
--- a/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.tsx
+++ b/airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.tsx
@@ -28,16 +28,18 @@ type Props = {
readonly dagRuns?: Array<DagRunAssetReference>;
};
-export const TriggeredRuns = ({ dagRuns }: Props) => {
+const DagRunGroup = ({
+ dagRuns,
+ prefix,
+}: {
+ readonly dagRuns: Array<DagRunAssetReference>;
+ readonly prefix: string;
+}) => {
const { t: translate } = useTranslation("common");
- if (dagRuns === undefined || dagRuns.length === 0) {
- return undefined;
- }
-
return dagRuns.length === 1 ? (
<Flex flexWrap="wrap" gap={1}>
- <Text flexShrink={0}>{`${translate("triggered")}
${translate("dagRun_one")}`}: </Text>
+ <Text flexShrink={0}>{`${prefix} ${translate("dagRun_one")}`}: </Text>
<StateBadge state={dagRuns[0]?.state as DagRunState} />
<RouterLink overflowWrap="anywhere"
to={`/dags/${dagRuns[0]?.dag_id}/runs/${dagRuns[0]?.run_id}`}>
{dagRuns[0]?.dag_id}
@@ -48,14 +50,14 @@ export const TriggeredRuns = ({ dagRuns }: Props) => {
<Popover.Root autoFocus={false} lazyMount unmountOnExit>
<Popover.Trigger asChild>
<Button variant="outline">
- {`${dagRuns.length} ${translate("triggered")}
${translate("dagRun_other", { count: dagRuns.length })}`}
+ {`${dagRuns.length} ${prefix} ${translate("dagRun_other", { count:
dagRuns.length })}`}
</Button>
</Popover.Trigger>
<Popover.Content css={{ "--popover-bg": "colors.bg.emphasized" }}
width="fit-content">
<Popover.Arrow />
<Popover.Body>
{dagRuns.map((dagRun) => (
- <Flex gap={1} key={dagRun.dag_id} my={2}>
+ <Flex gap={1} key={`${dagRun.dag_id}-${dagRun.run_id}`} my={2}>
<StateBadge state={dagRun.state as DagRunState} />
<RouterLink
to={`/dags/${dagRun.dag_id}/runs/${dagRun.run_id}`}>{dagRun.dag_id}</RouterLink>
</Flex>
@@ -65,3 +67,27 @@ export const TriggeredRuns = ({ dagRuns }: Props) => {
</Popover.Root>
);
};
+
+export const TriggeredRuns = ({ dagRuns }: Props) => {
+ const { t: translate } = useTranslation("common");
+
+ if (dagRuns === undefined || dagRuns.length === 0) {
+ return undefined;
+ }
+
+ // An asset event is linked to every run that consumed it, but only a run's
most recent consumed
+ // event triggered it (backend flag). The rest were merely included, so
label them accordingly.
+ const triggeredRuns = dagRuns.filter((dagRun) => dagRun.triggering);
+ const includedRuns = dagRuns.filter((dagRun) => !dagRun.triggering);
+
+ return (
+ <Flex direction="column" gap={1}>
+ {triggeredRuns.length > 0 ? (
+ <DagRunGroup dagRuns={triggeredRuns} prefix={translate("triggered")} />
+ ) : undefined}
+ {includedRuns.length > 0 ? (
+ <DagRunGroup dagRuns={includedRuns} prefix={translate("includedIn")} />
+ ) : undefined}
+ </Flex>
+ );
+};
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 462d1605af0..80c482ca245 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -832,7 +832,7 @@ class TestGetAssetEvents(TestAssets):
session.commit()
assert len(assets) == 2
- with assert_queries_count(3):
+ with assert_queries_count(4):
response = test_client.get("/assets/events")
assert response.status_code == 200
@@ -861,6 +861,7 @@ class TestGetAssetEvents(TestAssets):
"data_interval_start":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"data_interval_end":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
+ "triggering": True,
}
],
"timestamp":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
@@ -890,6 +891,7 @@ class TestGetAssetEvents(TestAssets):
"data_interval_start":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"data_interval_end":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
+ "triggering": True,
}
],
"timestamp":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
@@ -899,6 +901,34 @@ class TestGetAssetEvents(TestAssets):
"total_entries": 2,
}
+ def test_only_most_recent_consumed_event_is_flagged_as_triggering(self,
test_client, session):
+ """A run consuming several events marks only its most recent consumed
event as triggering it."""
+ self.create_assets(num=1)
+ older_event = AssetEvent(id=1, asset_id=1, extra={},
timestamp=DEFAULT_DATE)
+ newer_event = AssetEvent(id=2, asset_id=1, extra={},
timestamp=DEFAULT_DATE + timedelta(days=1))
+ session.add_all([older_event, newer_event])
+ dag_run = DagRun(
+ dag_id="source_dag_id",
+ run_id="run_1",
+ run_type=DagRunType.MANUAL,
+ logical_date=DEFAULT_DATE + timedelta(days=1),
+ start_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=DagRunState.SUCCESS,
+ )
+ dag_run.end_date = DEFAULT_DATE
+ session.add(dag_run)
+ session.flush()
+ dag_run.consumed_asset_events.extend([older_event, newer_event])
+ session.commit()
+
+ response = test_client.get("/assets/events")
+
+ assert response.status_code == 200
+ events = {event["id"]: event for event in
response.json()["asset_events"]}
+ assert events[1]["created_dagruns"][0]["triggering"] is False
+ assert events[2]["created_dagruns"][0]["triggering"] is True
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get("/assets/events")
assert response.status_code == 401
@@ -1049,6 +1079,7 @@ class TestGetAssetEvents(TestAssets):
"data_interval_start":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"data_interval_end":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
+ "triggering": True,
}
],
"timestamp":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
@@ -1078,6 +1109,7 @@ class TestGetAssetEvents(TestAssets):
"data_interval_start":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"data_interval_end":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
"partition_key": None,
+ "triggering": True,
}
],
"timestamp":
from_datetime_to_zulu_without_ms(DEFAULT_DATE),
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index d574fc58637..9d27ddf2817 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -2013,7 +2013,7 @@ class TestGetDagRunAssetTriggerEvents:
session.commit()
assert event.timestamp
- with assert_queries_count(3):
+ with assert_queries_count(4):
response = test_client.get(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents",
)
@@ -2043,6 +2043,7 @@ class TestGetDagRunAssetTriggerEvents:
"start_date":
from_datetime_to_zulu_without_ms(dr.start_date),
"state": "running",
"partition_key": partition_key,
+ "triggering": True,
}
],
"partition_key": partition_key,
@@ -2052,6 +2053,58 @@ class TestGetDagRunAssetTriggerEvents:
}
assert response.json() == expected_response
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ @pytest.mark.parametrize("num_events", [1, 5])
+ def test_query_count_does_not_scale_with_consumed_events(
+ self, num_events, test_client, dag_maker, session
+ ):
+ """A run consuming N asset events must not issue a query per event
(guard against N+1)."""
+ asset1 = Asset(name="ds1", uri="file:///da1")
+ with dag_maker(
+ dag_id="source_dag", start_date=START_DATE1,
schedule=timedelta(days=1), session=session
+ ):
+ EmptyOperator(task_id="task", outlets=[asset1])
+ source_run = dag_maker.create_dagrun()
+ ti = source_run.task_instances[0]
+ asset1_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset1.uri))
+
+ events = [
+ AssetEvent(
+ asset_id=asset1_id,
+ source_task_id=ti.task_id,
+ source_dag_id=ti.dag_id,
+ source_run_id=ti.run_id,
+ source_map_index=ti.map_index,
+ timestamp=START_DATE1 + timedelta(minutes=index),
+ )
+ for index in range(num_events)
+ ]
+ session.add_all(events)
+
+ with dag_maker(
+ dag_id="TEST_DAG_ID", start_date=START_DATE1,
schedule=timedelta(days=1), session=session
+ ):
+ pass
+ consuming_run = dag_maker.create_dagrun(run_id="TEST_DAG_RUN_ID",
run_type=DagRunType.ASSET_TRIGGERED)
+ for event in events:
+ consuming_run.consumed_asset_events.append(event)
+ session.commit()
+
+ # Constant regardless of the number of consumed events (parametrized 1
vs 5).
+ with assert_queries_count(4):
+ response =
test_client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents")
+
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["total_entries"] == num_events
+ # Only the run's most recent consumed event triggered it; the rest
were merely included.
+ triggering_by_event = {
+ event["id"]: event["created_dagruns"][0]["triggering"] for event
in payload["asset_events"]
+ }
+ newest_event_id = max(event.id for event in events)
+ assert triggering_by_event[newest_event_id] is True
+ assert all(value is False for eid, value in
triggering_by_event.items() if eid != newest_event_id)
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents",
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 3e9681b4c18..9f464e8bda3 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -533,6 +533,13 @@ class DagRunAssetReference(BaseModel):
data_interval_start: Annotated[datetime | None, Field(title="Data Interval
Start")]
data_interval_end: Annotated[datetime | None, Field(title="Data Interval
End")]
partition_key: Annotated[str | None, Field(title="Partition Key")]
+ triggering: Annotated[
+ bool,
+ Field(
+ description="Whether this asset event triggered the referenced dag
run. Only a run's most recent consumed asset event triggers it; earlier
consumed events are included in the run but did not trigger it.",
+ title="Triggering",
+ ),
+ ]
class DagRunMutableStates(str, Enum):
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index 52fa594afd5..aaad996ec99 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -370,6 +370,7 @@ class TestAssetsOperations:
data_interval_start=datetime.datetime(2025, 1, 1, 0, 0, 0),
data_interval_end=datetime.datetime(2025, 1, 1, 0, 0, 0),
partition_key=None,
+ triggering=True,
)
asset_event_response = AssetEventResponse(
