This is an automated email from the ASF dual-hosted git repository.
bbovenzi pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 15b6495a9dd Mark only a run's most recent asset event as triggering it
(#71441) (#71547)
15b6495a9dd is described below
commit 15b6495a9dd74de6012303c7e8b5f3d25b46ca0b
Author: Pierre Jeambrun <[email protected]>
AuthorDate: Thu Aug 13 17:01:28 2026 +0200
Mark only a run's most recent asset event as triggering it (#71441) (#71547)
* 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
(cherry picked from commit 3f66d0f2da5c81e010ea280b24f47175cfef02f4)
# Conflicts:
# airflow-core/src/airflow/api_fastapi/common/db/assets.py
# airflow-core/src/airflow/ui/src/components/Assets/TriggeredRuns.tsx
---
.../src/airflow/api_fastapi/common/db/assets.py | 68 +++++++++++++++++
.../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 | 5 +-
.../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 +
uv.lock | 4 +-
16 files changed, 364 insertions(+), 17 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
new file mode 100644
index 00000000000..741a7c17591
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/db/assets.py
@@ -0,0 +1,68 @@
+# 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 Iterable
+from typing import TYPE_CHECKING
+
+from sqlalchemy import func, select
+
+from airflow.models.asset import AssetEvent, association_table
+from airflow.models.dagrun import DagRun
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+
+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 3be0707bee1..98d4667449c 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
@@ -141,6 +141,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 34068eab85b..f55f87f5567 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
@@ -13544,6 +13544,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:
@@ -13556,6 +13562,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 9db8e1bffbd..b3c41514b7d 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
@@ -70,6 +70,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
@@ -351,10 +352,10 @@ def get_asset_events(
assets_event_select = assets_event_select.options(
subqueryload(AssetEvent.created_dagruns), joinedload(AssetEvent.asset)
)
- assets_events = session.scalars(assets_event_select)
+ 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 19bffb1015c..4b8cdbedb2d 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
@@ -98,6 +98,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,
@@ -277,7 +278,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(
@@ -286,7 +290,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 73fd78317ed..bb7983cc721 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
@@ -4173,11 +4173,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 975c7cc99bc..bd476adaf81 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
@@ -1069,6 +1069,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 70e600b332d..f24027404b0 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
@@ -150,6 +150,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 de061c49b39..d73623ed38e 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 gap={1}>
- <Text>{`${translate("triggered")} ${translate("dagRun_one")}`}: </Text>
+ <Text>{`${prefix} ${translate("dagRun_one")}`}: </Text>
<StateBadge state={dagRuns[0]?.state as DagRunState} />
<RouterLink
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 b4a4e697f62..29327d1d404 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
@@ -824,7 +824,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
@@ -853,6 +853,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),
@@ -882,6 +883,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),
@@ -891,6 +893,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
@@ -1041,6 +1071,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),
@@ -1070,6 +1101,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 57fb84cd6b2..6c633fe8a2a 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
@@ -1897,7 +1897,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",
)
@@ -1927,6 +1927,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,
@@ -1936,6 +1937,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 60a6cd33c96..3cb6bcc83f4 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -500,6 +500,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 b57883f7c42..334e5993560 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -325,6 +325,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(
diff --git a/uv.lock b/uv.lock
index 8edd29b6740..18913ff3353 100644
--- a/uv.lock
+++ b/uv.lock
@@ -976,7 +976,7 @@ wheels = [
[[package]]
name = "apache-airflow"
-version = "3.3.0"
+version = "3.3.1"
source = { editable = "." }
dependencies = [
{ name = "apache-airflow-core" },
@@ -1913,7 +1913,7 @@ requires-dist = [
[[package]]
name = "apache-airflow-core"
-version = "3.3.0"
+version = "3.3.1"
source = { editable = "airflow-core" }
dependencies = [
{ name = "a2wsgi" },
