Lee-W commented on code in PR #66030:
URL: https://github.com/apache/airflow/pull/66030#discussion_r3371233020
##########
airflow-core/src/airflow/partition_mappers/temporal.py:
##########
@@ -419,3 +422,130 @@ def normalize(self, dt: datetime) -> datetime:
second=0,
microsecond=0,
)
+
+
+# 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``).
+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())
+ """
+
+ 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__)
Review Comment:
We don't. The built-in windows cover enough cases, and anyone can write
their own parition mapper without going through our Window. It also raise
`ValueError` at construction time if user's not providing the right Window. I'd
rather keep it as-is until there's an actual user who asks.
--
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]
