Add FanOutMapper for one-to-many partition fan-out

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",

.pre-commit-config.yaml

@@ -248,6 +248,16 @@ repos:
  files: ^\.github/workflows/ci-(arm|amd)\.yml$
  pass_filenames: false
  require_serial: true
+ - id: check-partition-mapper-defaults-in-sync
+ name: Check FanOutMapper default mapper table stays in sync (core/SDK)
+ entry: ./scripts/ci/prek/check_partition_mapper_defaults_in_sync.py
+ language: python
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/partition_mappers/temporal\.py$|
+ ^task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal\.py$
+ pass_filenames: false
+ require_serial: true
  - id: sync-uv-min-version-markers
  name: Sync `# sync-uv-min-version` markers with [tool.uv] required-version
  entry: ./scripts/ci/prek/sync_uv_min_version_markers.py

airflow-core/newsfragments/66030.feature.rst

@@ -0,0 +1 @@
+Add ``FanOutMapper`` for one-to-many partition mapping (e.g. one weekly upstream key to seven daily downstream Dag runs). It composes ``upstream_mapper + window + downstream_mapper``, mirroring the shape of ``RollupMapper`` and reusing the existing ``Window`` classes (``DayWindow``, ``WeekWindow``, ``MonthWindow``, ``QuarterWindow``, ``YearWindow``). A new ``[scheduler] partition_mapper_max_downstream_keys`` config caps the number of downstream keys produced per upstream event by any ``PartitionMapper`` (built-in or custom). When the cap is exceeded, no Dag runs are queued for that upstream event and a ``Log`` row with ``event="partition fan-out exceeded"`` is written against the source task instance, recording the asset, target Dag, key count, and cap value.

airflow-core/src/airflow/assets/manager.py

@@ -552,6 +552,8 @@ def _queue_partitioned_dags(
  )
  return
 
+ max_downstream_keys = conf.getint("scheduler", "partition_mapper_max_downstream_keys")
+
  for target_dag in partition_dags:
  if TYPE_CHECKING:
  assert partition_key is not None
@@ -600,15 +602,36 @@ def _queue_partitioned_dags(
  continue
 
  if is_container(target_key):
- # TODO (AIP-76): This never happens now. When we implement
- # one-to-many partition key mapping, this should also add a
- # config to cap the iterable size so the scheduler does not
- # blow up with an incorrectly implemented PartitionMapper.
- target_keys: Iterable[str] = target_key
+ target_keys: list[str] = list(target_key)
  else:
  target_keys = [target_key]
  del target_key
 
+ if len(target_keys) > max_downstream_keys:
+ log.error(
+ "Partition mapper produced more downstream keys than allowed; skipping queue.",
+ asset_id=asset_id,
+ source_partition_key=partition_key,
+ target_dag=target_dag.dag_id,
+ produced_keys=len(target_keys),
+ max_downstream_keys=max_downstream_keys,
+ )
+ session.add(
+ Log(
+ event="partition fan-out exceeded",
+ extra=(
+ f"Partition mapper for asset (name='{asset_model.name}', "
+ f"uri='{asset_model.uri}') in target Dag '{target_dag.dag_id}' "
+ f"produced {len(target_keys)} downstream keys from "
+ f"partition_key='{partition_key}', exceeding "
+ f"[scheduler] partition_mapper_max_downstream_keys={max_downstream_keys}. "
+ f"No Dag runs were queued for this event."
+ ),
+ task_instance=task_instance,
+ )
+ )
+ continue
+
  for target_key in target_keys:
  apdr = cls._get_or_create_apdr(
  target_key=target_key,

airflow-core/src/airflow/config_templates/config.yml

@@ -2688,6 +2688,19 @@ scheduler:
  type: integer
  default: "20"
  see_also: ":ref:`scheduler:ha:tunables`"
+ partition_mapper_max_downstream_keys:
+ description: |
+ Maximum number of downstream partition keys a single ``PartitionMapper``
+ invocation may produce. When any partition mapper (built-in or custom)
+ expands one upstream key into more keys than this limit, the scheduler
+ skips queuing the runs for that asset event and logs an error against
+ the source task instance. This guards against a misconfigured
+ ``PartitionMapper`` from queuing an unbounded number of Dag runs per
+ upstream event.
+ version_added: 3.3.0
+ type: integer
+ example: ~
+ default: "1000"
  use_job_schedule:
  description: |
  Turn off scheduler use of cron intervals by setting this to ``False``.

airflow-core/src/airflow/example_dags/example_asset_partition.py

@@ -25,6 +25,7 @@
  Asset,
  CronPartitionTimetable,
  DayWindow,
+ FanOutMapper,
  IdentityMapper,
  MonthWindow,
  PartitionAtRuntime,
@@ -34,7 +35,9 @@
  StartOfDayMapper,
  StartOfHourMapper,
  StartOfMonthMapper,
+ StartOfWeekMapper,
  StartOfYearMapper,
+ WeekWindow,
  asset,
  task,
 )
@@ -357,3 +360,52 @@ def summarise_team_a_month(dag_run=None):
  print(f"All daily partitions received. Month: {dag_run.partition_key}")
 
  summarise_team_a_month()
+
+
+# --- Fan-out: one weekly upstream → seven daily downstream Dag runs ----------
+
+weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")
+
+
+with DAG(
+ dag_id="train_weekly_model",
+ schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
+ catchup=False,
+ tags=["example", "model", "training"],
+):
+ """Train a weekly model artifact every Monday at 00:00 UTC."""
+
+ @task(outlets=[weekly_model_artifact])
+ def train_model():
+ """Materialize the model artifact for the current weekly partition."""
+ pass
+
+ train_model()
+
+
+with DAG(
+ dag_id="daily_inference",
+ schedule=PartitionedAssetTimetable(
+ assets=weekly_model_artifact,
+ # FanOutMapper composes upstream_mapper + window + (optional) downstream_mapper.
+ # WeekWindow.to_upstream() yields seven daily datetimes inside one week,
+ # and the default downstream_mapper for WeekWindow is StartOfDayMapper, so
+ # a weekly upstream key fans out to seven ``%Y-%m-%d`` downstream keys.
+ default_partition_mapper=FanOutMapper(
+ upstream_mapper=StartOfWeekMapper(),
+ window=WeekWindow(),
+ ),
+ ),
+ catchup=False,
+ tags=["example", "model", "inference"],
+):
+ """Run daily inference, fanning the weekly model artifact out to one Dag run per day."""
+
+ @task
+ def run_inference(dag_run=None):
+ """Run inference for one daily partition derived from the weekly model."""
+ if TYPE_CHECKING:
+ assert dag_run
+ print(dag_run.partition_key)
+
+ run_inference()

airflow-core/src/airflow/partition_mappers/temporal.py

@@ -18,15 +18,18 @@
 
 import re
 from abc import abstractmethod
+from collections.abc import Iterable
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from airflow._shared.timezones.timezone import make_aware, parse_timezone
 from airflow.partition_mappers.base import PartitionMapper
 
 if TYPE_CHECKING:
  from pendulum import FixedTimezone, Timezone
 
+ from airflow.partition_mappers.window import Window
+
 
 _STRPTIME_PATTERNS: dict[str, str] = {
  "%Y": r"\d{4}",
@@ -419,3 +422,132 @@ def normalize(self, dt: datetime) -> datetime:
  second=0,
  microsecond=0,
  )
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream keys.
+
+ Compose an ``upstream_mapper`` (parses the coarse upstream key and
+ normalizes it to its period start) with a ``window`` (enumerates the
+ members of that period). ``downstream_mapper`` formats each member into a
+ downstream key string; if omitted, a default is chosen from the window
+ class.
+
+ ``downstream_mapper`` must be passed explicitly for any window without an
+ entry in the default table — currently ``HourWindow`` and any custom
+ ``Window`` subclass. Constructing a ``FanOutMapper`` for those windows
+ without a ``downstream_mapper`` raises ``ValueError`` at Dag-parse time.
+
+ Symmetric to :class:`~airflow.partition_mappers.base.RollupMapper`: rollup
+ is N→1 (downstream waits until all members arrive), fan-out is 1→N (one
+ upstream event creates one downstream Dag run per member).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ # Keep ``FanOutMapper.default_downstream_mapper_by_window_name`` in sync with
+ # the SDK copy in
+ # ``task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py`` —
+ # the SDK and core class hierarchies are independent (the SDK cannot import
+ # core), so both sides carry the same defaults and the lookup is by class
+ # name. When adding a new ``Window`` subclass, extend the table on both
+ # sides; a missing entry raises ``ValueError`` at ``FanOutMapper.__init__``
+ # (see ``FanOutMapper._resolve_default_downstream_mapper``). The
+ # ``check-partition-mapper-defaults-in-sync`` prek hook enforces that the
+ # two tables stay identical.
+ default_downstream_mapper_by_window_name: ClassVar[dict[str, type[_BaseTemporalMapper]]] = {
+ "DayWindow": StartOfHourMapper,
+ "WeekWindow": StartOfDayMapper,
+ "MonthWindow": StartOfDayMapper,
+ "QuarterWindow": StartOfMonthMapper,
+ "YearWindow": StartOfMonthMapper,
+ }
+
+ @classmethod
+ def _resolve_default_downstream_mapper(cls, window: Window) -> PartitionMapper:
+ """
+ Return the conventional downstream mapper for *window*.
+
+ Looked up by the window's class **name** rather than identity so that
+ the SDK ``Window`` classes (used in Dag-author code) and the core
+ ``Window`` classes (used after deserialization) both resolve to the
+ same default. Subclasses can extend or override the defaults by
+ setting :attr:`default_downstream_mapper_by_window_name` on the
+ subclass.
+ """
+ mapper_cls = cls.default_downstream_mapper_by_window_name.get(type(window).__name__)
+ if mapper_cls is None:
+ raise ValueError(
+ f"{cls.__name__} has no default downstream_mapper for window type "
+ f"{type(window).__name__}; pass downstream_mapper explicitly."
+ )
+ return mapper_cls()
+
+ def __init__(
+ self,
+ *,
+ upstream_mapper: PartitionMapper,
+ window: Window,
+ downstream_mapper: PartitionMapper | None = None,
+ ) -> None:
+ self.upstream_mapper = upstream_mapper
+ self.window = window
+ self.downstream_mapper = downstream_mapper or self._resolve_default_downstream_mapper(window)
+
+ def to_downstream(self, key: str) -> Iterable[str]:
+ # Round-trip the upstream key through its mapper to obtain the
+ # period-start datetime (decoded form). This keeps the upstream_mapper
+ # opaque — we don't need to know whether it's temporal or segment.
+ formatted = self.upstream_mapper.to_downstream(key)
+ if not isinstance(formatted, str):
+ raise TypeError(
+ "FanOutMapper.upstream_mapper must produce a single key from "
+ "to_downstream; chained fan-out (mapper that itself returns multiple keys) "
+ "is not supported."
+ )
+ coarse = self.upstream_mapper.decode_downstream(formatted)
+ return [_format_with(self.downstream_mapper, item) for item in self.window.to_upstream(coarse)]
+
+ def serialize(self) -> dict[str, Any]:
+ from airflow.serialization.encoders import encode_partition_mapper, encode_window
+
+ return {
+ "upstream_mapper": encode_partition_mapper(self.upstream_mapper),
+ "window": encode_window(self.window),
+ "downstream_mapper": encode_partition_mapper(self.downstream_mapper),
+ }
+
+ @classmethod
+ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
+ from airflow.serialization.decoders import decode_partition_mapper, decode_window
+
+ return cls(
+ upstream_mapper=decode_partition_mapper(data["upstream_mapper"]),
+ window=decode_window(data["window"]),
+ downstream_mapper=decode_partition_mapper(data["downstream_mapper"]),
+ )
+
+
+def _format_with(mapper: PartitionMapper, decoded: Any) -> str:
+ """
+ Format *decoded* using *mapper*'s downstream format.
+
+ Three-layer fallback, in order:
+
+ 1. *mapper* exposes a callable ``format`` attribute (all temporal mappers do)
+ — delegates to ``mapper.format(decoded)``, which uses ``output_format``.
+ 2. *decoded* is a ``datetime`` instance — returns ``decoded.isoformat()``,
+ producing a stable ISO-8601 string that round-trips via
+ ``datetime.fromisoformat``.
+ 3. Anything else — ``str(decoded)`` as a last-resort fallback.
+ """
+ formatter = getattr(mapper, "format", None)
+ if callable(formatter):
+ return formatter(decoded)
+ if isinstance(decoded, datetime):
+ return decoded.isoformat()
+ return str(decoded)

airflow-core/src/airflow/serialization/encoders.py

@@ -42,6 +42,7 @@
  DeltaDataIntervalTimetable,
  DeltaTriggerTimetable,
  EventsTimetable,
+ FanOutMapper,
  HourWindow,
  IdentityMapper,
  MonthWindow,
@@ -433,6 +434,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
  BUILTIN_PARTITION_MAPPERS: dict[type, str] = {
  AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
  ChainMapper: "airflow.partition_mappers.chain.ChainMapper",
+ FanOutMapper: "airflow.partition_mappers.temporal.FanOutMapper",
  IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
  ProductMapper: "airflow.partition_mappers.product.ProductMapper",
  RollupMapper: "airflow.partition_mappers.base.RollupMapper",
@@ -499,6 +501,14 @@ def _(self, partition_mapper: RollupMapper) -> dict[str, Any]:
  "window": encode_window(partition_mapper.window),
  }
 
+ @serialize_partition_mapper.register
+ def _(self, partition_mapper: FanOutMapper) -> dict[str, Any]:
+ return {
+ "upstream_mapper": encode_partition_mapper(partition_mapper.upstream_mapper),
+ "window": encode_window(partition_mapper.window),
+ "downstream_mapper": encode_partition_mapper(partition_mapper.downstream_mapper),
+ }
+
  BUILTIN_WINDOWS: dict[type, str] = {
  HourWindow: "airflow.partition_mappers.window.HourWindow",
  DayWindow: "airflow.partition_mappers.window.DayWindow",