GitHub user Leondon9 created a discussion: Coverage-based completeness check
for partitioned assets (streaming producers)
> **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:
```text
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:
```python
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".
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @Leondon9 before posting
GitHub link: https://github.com/apache/airflow/discussions/74595
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
