[renat-sagut]

Asset-triggered Dags currently combine queued asset events into a single run. This PR makes one run per event the default and adds batch_asset_events to AssetTriggeredTimetable, PartitionedAssetTimetable, and AssetOrTimeSchedule so authors can opt into batching.

The new [scheduler] batch_asset_events setting defaults to False. An explicit timetable argument overrides it. Setting it to True and reparsing Dags restores the previous behavior; serialized timetables without the field retain batching until they are reparsed.

from airflow.sdk import DAG, Asset, AssetTriggeredTimetable, PartitionedAssetTimetable

source = Asset("s3://bucket/source")

with DAG(
    dag_id="per-event",
    schedule=AssetTriggeredTimetable(assets=source, batch_asset_events=False),
):
    ...

with DAG(
    dag_id="per-partition-event",
    schedule=PartitionedAssetTimetable(assets=source, batch_asset_events=False),
):
    ...

With batching disabled, partitioned events get separate pending runs for each mapped partition. Non-partitioned run creation respects max_active_runs and the scheduler's creation budget, retaining queued and catchup events until subsequent passes can consume them.

Multi-asset AND expressions and partition rollups require batch_asset_events=True. Parsing rejects those combinations when batching is disabled, preventing pending runs whose conditions can never be satisfied. The examples, documentation, and significant-change note explain this compatibility requirement.

Regression coverage includes configuration overrides, SDK/core serialization and legacy payloads, repeated partition keys, simultaneous event timestamps, bounded backlog draining, catchup history, and Dag parsing.

Related #53896

---

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

• Yes — CodeWhale and Codex (GPT-6)

Generated-by: CodeWhale and Codex (GPT-6) following the guidelines

[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]

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).

[vatsrahul1001]

@renat-sagut looks like we have a conflict can you resolve

[potiuk]

@renat-sagut I've removed the ready for maintainer review label because the next step here is yours: unresolved review thread(s) from @uranusjr awaiting your reply. It'll go back into the maintainer queue automatically once that's done — no need to re-add the label by hand. Next move: address the unresolved threads and reply to @uranusjr. There is no rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. If you have questions, an Apache Airflow maintainer — a real person — is happy to help. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.