Reduce scheduler memory for asset-triggered Dag runs

[viiccwen]

Description

closes #69848

Asset-triggered Dag run creation currently materializes every matching AssetEvent ORM object, including its JSON metadata, before writing the Dag run association rows. Large event windows can therefore cause scheduler memory spikes far beyond the raw payload size.

This change writes the association rows with a server-side INSERT ... SELECT while preserving the existing event predicates, transaction boundary, exact event membership, and relationship behavior.

Scheduler-side AssetEvent ORM and JSON payload materialization drops from O(events) to approximately O(1). The database still performs O(events) work to select event IDs and persist one association row per event; this change moves that set operation out of Python rather than eliminating it.

The partitioned-asset path uses the same set-based association helper. No schema or configuration change is needed.

Benchmark

The benchmark invokes the real scheduler path on SQLite, PostgreSQL, and MySQL with a 512-byte JSON
payload per event. Setup is outside the measured interval.

Database	Events	Duration (before/after)	Speedup	Python peak (before/after)	Peak reduction
SQLite	1,000	0.537 / 0.255s	2.11×	6.23 / 0.16 MB	97.2%
SQLite	10,000	1.785 / 0.024s	74.38×	54.24 / 0.11 MB	99.8%
SQLite	50,000	12.545 / 0.071s	176.69×	274.21 / 0.13 MB	99.95%
PostgreSQL	1,000	0.339 / 0.237s	1.43×	7.39 / 1.84 MB	75.1%
PostgreSQL	10,000	2.510 / 0.096s	26.15×	62.34 / 0.12 MB	99.8%
PostgreSQL	50,000	10.046 / 0.688s	14.60×	289.72 / 0.14 MB	99.95%
MySQL	1,000	0.161 / 0.032s	5.00×	5.80 / 0.16 MB	97.2%
MySQL	10,000	4.478 / 0.185s	24.21×	54.54 / 0.11 MB	99.8%
MySQL	50,000	21.502 / 1.977s	10.88×	272.63 / 0.13 MB	99.95%

The 1,000-event result includes first-call and tracing warm-up overhead.

Notes

• consumed_asset_event_count is designed for logging, can discuss leave it or not.

---

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

• Yes - Codex (GPT 5.6 Sol)

---

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

[viiccwen]

cc @Lee-W, @jason810496. ✨🙌

[Lee-W]

Thanks! I remember we did something similar elsewhere. the overall idea looks good to me