Support waiting for asset partition readiness from time-scheduled DAGs

[Leondon9]

I am looking for guidance before attempting an implementation.

In time-scheduled DW/DataLake DAGs, data_interval_start and data_interval_end are often still the primary processing contract. Asset-triggered DAGs are useful, but switching a heavy batch consumer to schedule=[Asset(...)] changes the run semantics because asset-triggered runs intentionally do not have a logical date or data interval.

The use case I am trying to model is:

• keep the consumer DAG time-scheduled, for example @daily
• keep SQL/business logic based on the consumer's data interval
• wait for an upstream Airflow asset partition, for example orders / dt=2026-05-21
• avoid coupling the consumer to the producer DAG's exact schedule via ExternalTaskSensor(execution_date_fn=...)

Today I can solve this with provider-specific sensors, such as S3, Hive, BigQuery, or Databricks partition sensors, or with ExternalTaskSensor. But I could not find an Airflow-native task-level way to wait for an AssetEvent / asset partition recorded in Airflow metadata.

A concrete shape might look like this:

orders = Asset("warehouse://orders")

wait_orders = AssetPartitionSensor(
    task_id="wait_orders",
    asset=orders,
    partition_key="{{ data_interval_start | ds }}",
    deferrable=True,
)

The intended semantics would be:

• the consumer DAG is still created by its time schedule
• the sensor waits for an Airflow asset event matching asset + partition_key
• the consumer can continue using its own data_interval_start / data_interval_end
• the producer DAG can be scheduled, manual, backfilled, or otherwise changed, as long as it emits the expected asset partition event

This is not intended to replace asset-triggered scheduling, and I am not proposing to derive data_interval for asset-triggered DAG runs. I am trying to clarify whether there is room for a time-scheduled DAG to wait on asset partition readiness as a task-level dependency.

This seems related to the problem discussed in #55489, where users need a common date/partition concept across scheduled, triggered, and asset-aware workflows. The maintainer comments there also point toward asset partitions / dag_run.partition_key as the more general direction.

Questions:

1. Is this gap already intended to be covered by asset partitions in Airflow 3.2+?
2. If not, would a task-level asset partition readiness sensor fit Airflow's direction?
3. Should such a feature live in the standard provider, core, or only be documented as a pattern using provider-specific sensors?
4. What semantics would be acceptable: latest event exists, any event exists, source-run-aware matching, or something else?

If this direction makes sense, I would be interested in helping with a small scoped follow-up, likely starting with docs or a design issue before any implementation PR.

[Leondon9]

Update — since I opened this, the building blocks landed, so I'll answer my own questions and link the
in-flight work.

Q1 (covered by asset partitions in 3.2+?) Partially. Asset partitioning — partition_key on asset
events and PartitionedAssetTimetable — is there from 3.2. But querying asset events by
partition_key from a task needed Execution API filters that landed later in #64610 / #64611 (3.4). So
the pieces a task-level sensor needs only became available recently.

Q2 (does a task-level readiness sensor fit Airflow's direction?) Looks like yes — it's being
implemented as a standard-provider sensor rather than pushed into core scheduling.

Q3 (standard provider, core, or just a doc pattern?) Standard provider. The sensor and its trigger
live in providers/standard; the only core touch is wiring in the triggerer so the trigger can read
asset events while deferred (a trigger has no DB/Execution-API access of its own).

Q4 (matching semantics?) Event-existence, with an optional time bound. Waiting on "any event with
this partition key" is enough when the key is unique per event (a timestamp), but for keys that repeat
across runs (a region code) an unbounded lookup matches a stale event immediately — so an after
bound scoped to the run's interval is needed. A more general form also filters on event extra, time
range, and a required event count.

Two implementations are in progress, both on top of #64610:

• Add asset partition sensor #67941 — AssetPartitionSensor, focused on the partition-readiness case above (plus the triggerer
  wiring for deferrable mode).
• Add AssetEventSensor to standard provider #70225 — AssetEventSensor, a more general asset-event sensor (extra / time-range filters,
  expected_count, process_result).

They overlap heavily and will likely converge into one; tracking both here so anyone following this
thread can find the actual work.

---

Drafted-by: Claude Code (Opus 4.8); reviewed by @Leondon9 before posting