Coverage-based completeness check for partitioned assets (streaming producers)

[Leondon9]

TL;DR Streaming producers publish many small, sometimes irregular, partitions. Downstream consumers usually ask one question: "Is [start, end) fully covered, and if not, which gaps are missing?" The current partition model (AIP-76) answers "have these exact keys arrived?" This proposes letting asset events carry an interval, so Airflow can answer the coverage question. I'd like feedback before writing any code.

The use case

Producer	Emits	Consumer wants
Spark Structured Streaming, 5-minute micro-batches	one asset event per batch (partition_key=…T10:05:00)	run hourly / daily aggregation only when the period is complete
Iceberg commit job, every 5 minutes	one event per snapshot commit	"is June 2025 fully loaded?" before a backfill or a report

Example question from a consumer task:

asset = A, start = 2025-06-01, end = 2025-07-01
→ covered: False
→ missing: [2025-06-14T03:10, 2025-06-14T03:25)

Why the current model does not express this

AIP-76 rollup (RollupMapper + Window + WaitPolicy, from #64571 and #66848) works on discrete keys:

1. RollupMapper.to_upstream(downstream_key) lists the exact keys one downstream period needs.
2. The scheduler compares that set with the received source_partition_key values via WaitPolicy.is_satisfied_by_keys(matched, expected).

That fits producers with fixed, aligned partitions. It breaks down for streaming:

• An event has no end. AssetEvent stores partition_key (a string), extra, and timestamp. A key like 2025-06-01T10:05:00 tells us when the slice starts, not when it ends. The window fills that gap by assuming a fixed unit (one minute for HourWindow).
• Batch boundaries drift. Trigger delays, restarts, and reprocessing (one 1-hour event replacing twelve 5-minute ones) produce keys that a fixed key list never matches.
• "Missing" is a range, not a key. Consumers want gaps as time ranges, which set difference over keys only approximates.

Proposal: let events describe the interval they cover

Each event says which [start, end) it covers. Completeness becomes interval arithmetic: sort, merge overlaps, and take the complement inside the requested range. That runs in O(n log n) and returns the gaps directly.

How an event could carry its interval:

Option	Example	Cost
a. Convention in extra	{"interval_start": "…T10:05", "interval_end": "…T10:10"}	No schema change; needs a documented convention
b. ISO 8601 interval as the key	partition_key="2025-06-01T10:05/2025-06-01T10:10"	No schema change; affects key parsing in mappers
c. New AssetEvent columns	interval_start, interval_end (indexed)	DB migration; efficient range queries

Where the check could live

A. Task-side sensor / operator (small)

Use the existing Execution API asset-event lookup from a task:

events = inlet_events[A].partition_key_regexp_pattern(r"^2025-06-")
intervals = sorted((e.extra["interval_start"], e.extra["interval_end"]) for e in events)
missing = complement(merge(intervals), within=(start, end))

• Works today with option (a), with no core change.
• Gaps: no key-range filter (>=, <) on the asset-event endpoints, so ranges are narrowed with a regex. Large ranges need pagination: a month of 5-minute events is about 8,640 rows. The asset must also be declared in the task's inlets.
• Natural home: a sensor in providers/standard.

B. Scheduler-side coverage rollup (large)

Let the scheduler create the downstream Dag run when the period is covered, the way RollupMapper does today for key sets.

Area	Today	Would need
Expected upstream	to_upstream() returns a key set	an interval
WaitPolicy	set comparison	interval union
PartitionedAssetKeyLog	source_partition_key string	interval columns, migration, index
UI progress	"12 / 24 received"	covered ratio and gap list
MinimumCount(n)	count of keys	needs new semantics

B changes the AIP-76 model itself, so it needs community agreement before implementation.

Relationship to the smaller step change

I have opened a separate, much smaller PR (#74594): an optional step on HourWindow, DayWindow, and WeekWindow, for example HourWindow(step=timedelta(minutes=5)). It keeps the key-set model and covers the common case of aligned, fixed-size producers.

This coverage idea is complementary. It is for irregular or reprocessed slices, and for consumers who need the gaps as time ranges.

Questions for discussion

1. Is "coverage of [start, end)" a use case others see with partitioned assets, especially for streaming producers?
2. Which representation fits best: extra convention, interval keys, or columns?
3. Should we start with A (a sensor) and only consider B after the sensor sees real use?
4. For A, is a key-range filter on the Execution API asset-event endpoints acceptable?
5. How should overlapping or superseding events (a reprocessed hour replacing twelve 5-minute events) be treated: union, or latest wins?

Feedback is welcome, including "this is out of scope for Airflow".

[Leondon9]

@uranusjr, could you take a look at this discussion when you have a moment?