batch asset events

[renat-sagut]

Add batch_asset_events parameter to asset-triggered timetables

Introduces batch_asset_events: bool = True on AssetTriggeredTimetable and
PartitionedAssetTimetable (both SDK and core versions). When set to False,
asset-triggered Dags create one DagRun per individual asset event instead of
batching all events into a single DagRun.

Behavior

batch_asset_events	Partitioned (APDR)	Non-partitioned (ADRQ)
True (default)	Events for same partition key share one APDR → one DagRun consuming all events	One DagRun per tick consuming all queued events
False	Each event creates its own APDR → one DagRun per event	One DagRun per event (run_after = event timestamp)

Usage

batch_asset_events defaults to True. Set to False for one DagRun per event.

Non-partitioned:
Dag(dag_id="per-event", schedule=AssetTriggeredTimetable(assets=asset_1, batch_asset_ events=False))

Partitioned:
Dag(dag_id="per-event", schedule=PartitionedAssetTimetable(assets=asset_1, batch_asset _events=False))

Related #53896

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: codewhale following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[jscheffl]

Early feedback while CI is running: Can you add some documentation with a very simple example to airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst?

Note for reference, this is related to Slack discussion in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1778258939480149

[renat-sagut]

Early feedback while CI is running: Can you add some documentation with a very simple example to airflow-core/docs/authoring-and-scheduling/asset-scheduling.rst?

Note for reference, this is related to Slack discussion in https://apache-airflow.slack.com/archives/C06K9Q5G2UA/p1778258939480149

Done

[jscheffl]

nit: Can you add a marker that this is added in 3.3.0?

Suggested change

	By default, when multiple asset events arrive for the same Dag between
	.. versionadded:: 3.3.0
	By default, when multiple asset events arrive for the same Dag between

[jscheffl]

Besides the comment nit I think this is a good solution to fix the problem raised in #53896 and most part of #56750 in my view. (in #56750 part 3 is a bug whereas any background is missing allowing a triage, so maybe this need either more evidence or should be closed).

But besides the positive discussions in Slack would like to have sone Asste Event experts additional opinion, mostly @uranusjr (alternatively @Lee-W / @ashb).