uranusjr commented on code in PR #66030:
URL: https://github.com/apache/airflow/pull/66030#discussion_r3278839657
##########
task-sdk/src/airflow/sdk/definitions/partition_mappers/temporal.py:
##########
@@ -65,3 +70,72 @@ class StartOfYearMapper(_BaseTemporalMapper):
"""Map a time-based partition key to year."""
default_output_format = "%Y"
+
+
+class FanOutMapper(PartitionMapper):
+ """
+ Partition mapper that fans one upstream key out into multiple downstream
keys.
+
+ Compose an ``upstream_mapper`` (which parses the coarse upstream key and
+ normalizes it to a period start) with a ``window`` that 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 (e.g. ``WeekWindow`` → ``StartOfDayMapper``).
+
+ ``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.sdk.RollupMapper`: rollup is N→1 (downstream
+ waits for all members), fanout is 1→N (one upstream event creates many
+ downstream Dag runs).
+
+ .. code-block:: python
+
+ # Weekly upstream → 7 daily downstream Dag runs
+ FanOutMapper(upstream_mapper=StartOfWeekMapper(), window=WeekWindow())
+ """
+
+ 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
_resolve_default_downstream_mapper(window)
Review Comment:
Should this resolve the default immediately? Or should it resolve only after
serialisation to avoid duplicating logic? This also reminds me how timetable
does not normalise timezone in str currently. We should have a unified approach
to this kind of parameters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
