Add AssetAndTimeSchedule timetable

[aaron-y-chen]

Closes: #58056

Why

Use case: #58056

How

1. Add AssetAndTimeSchedule

Add a timetable that combines a time-based timetable with an asset condition.

It behaves like a scheduled timetable, not an asset-triggered timetable: when it creates a run, the run type is still SCHEDULED.

2. Gate DagRun creation before creating the DagRun

The scheduler creates a scheduled DagRun only when both conditions are satisfied: the schedule time is due and the required asset condition is ready.

If assets are missing, no DagRun is created and next_dagrun_create_after stays on the pending slot, so the same slot can be retried later without creating placeholder DagRuns.

3. Keep asset-aware scheduler behavior consistent

Timetables expose asset-aware scheduling through asset_triggered and asset_gated behavior flags, following the existing periodic pattern. These flags are also mirrored on the Task SDK base timetable so custom timetables can opt in without scheduler type checks.

DagModel.dags_needing_dagruns uses these flags to route asset-triggered Dags separately from asset-gated scheduled Dags. Asset-gated Dags still flow through the standard scheduled-run creation path (_create_dag_runs).

When DagModel.timetable_asset_gated is set, the scheduler re-checks the timetable's asset_condition under ADRQ row locks before creating the scheduled DagRun. If the condition is still satisfied, the created run links and consumes the selected asset events. The ADRQ locking, event-provenance, and consumption helpers are shared with the asset-triggered path.

What

Here are my test DAGs:

upstream_asset_producer

import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.sdk.definitions.asset import Asset

my_asset = Asset("/my/example/asset")

with DAG(
    dag_id="upstream_asset_producer",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    schedule="0 * * * *",
    catchup=False,
    tags=["example", "upstream"],
) as upstream_dag:
    BashOperator(
        task_id="produce_asset",
        bash_command="echo 'Asset produced!'",
        outlets=[my_asset],
    )

downstream_asset_and_time_consumer

import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.sdk.definitions.asset import Asset
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.timetables.assets import AssetAndTimeSchedule

my_asset = Asset("/my/example/asset")

with DAG(
    dag_id="downstream_asset_and_time_consumer",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    schedule=AssetAndTimeSchedule(
        timetable=CronTriggerTimetable("4 * * * *", timezone="UTC"),
        assets=[my_asset],
    ),
    catchup=False,
    max_active_runs=1,
    tags=["example", "downstream", "asset-and-time"],
) as downstream_dag:
    BashOperator(
        task_id="consume_asset",
        bash_command="echo 'Asset condition and time condition were both met!'",
    )

From the screenshots, the downstream DAG runs as expected: it runs only after both the timetable and the assets are ready.

1. Upstream dag

   

2. Asset production time

   

3. Downstream, AssetAndTimeSchedule

   

---

Important

🛠️ Maintainer triage note for @nailo2c · by @potiuk · 2026-06-22 06:31 UTC

Your review threads from @jscheffl look addressed — please confirm this PR is ready for maintainer review confirmation:

• 1 thread(s) show your engagement (post-review commits and/or replies).

The ball is in your court — you've been assigned to this PR. Reply yes / ready (and mark the threads resolved) and a maintainer will pick it up from the queue.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[uranusjr]

We can probably do some refactoring to extract some common logic between the And and Or timetables into a mixin. This can be a different PR.

[aaron-y-chen]

Hi @uranusjr, I'd appreciate it if you could take another look when you have time.

[shahar1]

AI Review

LGTM — the asset-gated scheduling logic (no-placeholder-DagRun retry, ADRQ row-locked re-check, Task SDK flag mirroring without DB leakage, migration symmetry) is correct and well covered by tests across Core and the Task SDK. CI is green. Two minor nits inline, neither blocking.

Smaller observations

• airflow-core/src/airflow/jobs/scheduler_job_runner.py:2832 — the new shared _lock_queued_asset_records helper unconditionally adds .options(joinedload(AssetDagRunQueue.asset)). The pre-existing _create_dag_runs_asset_triggered call site (which used to inline this query without the join) never reads .asset, so it now pays for an extra JOIN/hydration on every asset-triggered scheduler pass for no benefit. Not a correctness issue, just a small efficiency regression from the refactor — worth gating the eager load behind a parameter if it matters at scale.
• airflow-core/tests/unit/timetables/test_assets_timetable.py:398,421-422,450-451 — test_infer_manual_data_interval_and, test_next_dagrun_info_and, and test_generate_run_id_and use DateTime.now() instead of time_machine (this repo's testing standard: "Use time_machine for time-dependent tests. Do not use datetime.now()"). This mirrors the existing sibling AssetOrTimeSchedule tests in the same file rather than introducing a new deviation, and the assertions are structural (isinstance(...)) so it isn't flaky today — but new code shouldn't reproduce the pattern.

---

This review was drafted by an AI-assisted tool and confirmed by an Airflow maintainer. The maintainer approving this PR has read the findings and signed off. If something feels off, please reply on the PR and a maintainer will follow up.

More on how Airflow handles maintainer review:
contributing-docs/05_pull_requests.rst.

[ashb]

@shahar1 I'd like @Lee-W to review this before we merge please.

[Lee-W]

I will review it early tomorrow. Thanks @ashb for the reminder and @aaron-y-chen for the ongoing effort!

[Lee-W]

a few nits. nothing major. I'm good with merging it.

[Lee-W]

also need to resolve conflcit