Streaming producers (Spark Structured Streaming micro-batches, Iceberg commit jobs) publish one partition every few minutes, but the temporal rollup windows expect a fixed key spacing:

A producer that emits one key every 5 minutes sends 12 keys per hour, so with HourWindow() and the default WaitForAll() the hourly Dag run waits forever. The workarounds are emitting fake per-minute events, using MinimumCount(12) (which no longer tells you which batch is missing), or a custom window plugin per deployment.

This adds an optional step to HourWindow, DayWindow and WeekWindow:

from datetime import timedelta
from airflow.sdk import HourWindow, PartitionedAssetTimetable, RollupMapper, StartOfHourMapper

schedule = PartitionedAssetTimetable(
    assets=orders_5min,
    default_partition_mapper=RollupMapper(
        upstream_mapper=StartOfHourMapper(),
        window=HourWindow(step=timedelta(minutes=5)),
    ),
)
# Expected keys for "2026-10-06T10": 10:00:00, 10:05:00, …, 10:55:00 (12)

Behaviour

• to_upstream walks the period in step increments and reuses _build_directional_steps, so Window.Direction.BACKWARD keeps working.
• Validated at Dag parse time (SDK and core): step must be a positive timedelta with whole seconds that divides the period evenly (HourWindow(step=timedelta(minutes=7)) raises ValueError). MonthWindow, QuarterWindow and YearWindow reject step with TypeError, since their periods vary in length.
• FanOutMapper requires an explicit downstream_mapper when its window has a custom step, because the built-in defaults assume the window's default unit and would collapse several members onto one key.
• step=None keeps the current behaviour. Serialization omits step when it is None, so existing Dags keep their rollup fingerprint and pending partition Dag runs are not reset on upgrade. No DB migration; step lives in the serialized Dag JSON.
• The UI received/required counts come from to_upstream, so they show "12 / 12" without UI changes.

Out of scope / open questions

• Bucket origins other than the period start (e.g. 7-minute buckets crossing hour boundaries).
• Off-grid keys: a producer that sends 10:03 is logged but never matches. Should ingestion warn?
• Size cap: WeekWindow(step=timedelta(minutes=1)) expects 10,080 keys per pending partition Dag run. Should there be a limit similar to max_downstream_keys?
• Coverage-based completeness for irregular slices (events that carry their own [start, end)) is a larger change to the AIP-76 model; it is proposed separately in Coverage-based completeness check for partitioned assets (streaming producers) #74595 rather than folded into this PR.

Testing

• airflow-core/tests/unit/partition_mappers (390 passed), task-sdk/tests/task_sdk/definitions/test_partition_mappers.py (49 passed), serialization tests for window/partition/rollup/fingerprint, and scheduler rollup/partition tests (-k "rollup or partition", 51 passed).
• prek pre-commit and manual stages, including check-window-in-sync, mypy-airflow-core and mypy-task-sdk.
• breeze testing task-sdk-tests, breeze testing core-tests (DB on SQLite and non-DB, test types API/Always/CLI/Core/Other/Serialization) and breeze build-docs apache-airflow task-sdk pass locally. The provider suites selected by breeze verify were not run locally (memory limits); they are left to CI.

related: #64571, #66848, #69456

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

• Yes — Claude Code (Opus 5.5)

Generated-by: Claude Code (Opus 5.5) following the guidelines