refactor(partition-mapper): rename ToXXXMapper to StartOfXXXMapper

airflow-core/docs/authoring-and-scheduling/assets.rst

@@ -434,13 +434,13 @@ For downstream partition-aware scheduling, use ``PartitionedAssetTimetable``:
 
 .. code-block:: python
 
-    from airflow.sdk import DAG, HourlyMapper, PartitionedAssetTimetable
+    from airflow.sdk import DAG, StartOfHourMapper, PartitionedAssetTimetable
 
     with DAG(
         dag_id="clean_and_combine_player_stats",
         schedule=PartitionedAssetTimetable(
             assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
-            default_partition_mapper=HourlyMapper(),
+            default_partition_mapper=StartOfHourMapper(),
         ),
         catchup=False,
     ):
@@ -458,17 +458,17 @@ Partition mappers define how upstream partition keys are transformed to the
 downstream Dag partition key:
 
 * ``IdentityMapper`` keeps keys unchanged.
-* Temporal mappers such as ``HourlyMapper``, ``DailyMapper``, and
-  ``YearlyMapper`` normalize time keys to a chosen grain. For input key
+* Temporal mappers such as ``StartOfHourMapper``, ``StartOfDayMapper``, and
+  ``StartOfYearMapper`` normalize time keys to a chosen grain. For input key
   ``2026-03-10T09:37:51``, the default outputs are:
 
-  * ``HourlyMapper`` -> ``2026-03-10T09``
-  * ``DailyMapper`` -> ``2026-03-10``
-  * ``YearlyMapper`` -> ``2026``
+  * ``StartOfHourMapper`` -> ``2026-03-10T09``
+  * ``StartOfDayMapper`` -> ``2026-03-10``
+  * ``StartOfYearMapper`` -> ``2026``
 * ``ProductMapper`` maps composite keys segment-by-segment.
   It applies one mapper per segment and then rejoins the mapped segments.
   For example, with key ``us|2026-03-10T09:00:00``,
-  ``ProductMapper(IdentityMapper(), DailyMapper())`` produces
+  ``ProductMapper(IdentityMapper(), StartOfDayMapper())`` produces
   ``us|2026-03-10``.
 * ``AllowedKeyMapper`` validates that keys are in a fixed allow-list and
   passes the key through unchanged if valid.
@@ -481,10 +481,10 @@ Example of per-asset mapper configuration and composite-key mapping:
 
     from airflow.sdk import (
         Asset,
-        DailyMapper,
         IdentityMapper,
         PartitionedAssetTimetable,
         ProductMapper,
+        StartOfDayMapper,
     )
 
     regional_sales = Asset(uri="file://incoming/sales/regional.csv", name="regional_sales")
@@ -493,7 +493,7 @@ Example of per-asset mapper configuration and composite-key mapping:
         dag_id="aggregate_regional_sales",
         schedule=PartitionedAssetTimetable(
             assets=regional_sales,
-            default_partition_mapper=ProductMapper(IdentityMapper(), DailyMapper()),
+            default_partition_mapper=ProductMapper(IdentityMapper(), StartOfDayMapper()),
         ),
     ):
         ...
@@ -503,7 +503,7 @@ You can also override mappers for specific upstream assets with
 
 .. code-block:: python
 
-    from airflow.sdk import Asset, DAG, DailyMapper, IdentityMapper, PartitionedAssetTimetable
+    from airflow.sdk import Asset, DAG, StartOfDayMapper, IdentityMapper, PartitionedAssetTimetable
 
     hourly_sales = Asset(uri="file://incoming/sales/hourly.csv", name="hourly_sales")
     daily_targets = Asset(uri="file://incoming/sales/targets.csv", name="daily_targets")
@@ -513,7 +513,7 @@ You can also override mappers for specific upstream assets with
         schedule=PartitionedAssetTimetable(
             assets=hourly_sales & daily_targets,
             # Default behavior: map timestamp-like keys to daily keys.
-            default_partition_mapper=DailyMapper(),
+            default_partition_mapper=StartOfDayMapper(),
             # Override for assets that already emit daily partition keys.
             partition_mapper_config={
                 daily_targets: IdentityMapper(),

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

@@ -27,9 +27,9 @@
     IdentityMapper,
     PartitionedAssetTimetable,
     ProductMapper,
-    ToDailyMapper,
-    ToHourlyMapper,
-    ToYearlyMapper,
+    StartOfDayMapper,
+    StartOfHourMapper,
+    StartOfYearMapper,
     asset,
     task,
 )
@@ -77,15 +77,15 @@ def team_c_player_stats():
     dag_id="clean_and_combine_player_stats",
     schedule=PartitionedAssetTimetable(
         assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
-        default_partition_mapper=ToHourlyMapper(),
+        default_partition_mapper=StartOfHourMapper(),
     ),
     catchup=False,
     tags=["player-stats", "cleanup"],
 ):
     """
     Combine hourly partitions from Team A, B and C into a single curated dataset.
 
-    This Dag demonstrates multi-asset partition alignment using ToHourlyMapper.
+    This Dag demonstrates multi-asset partition alignment using StartOfHourMapper.
     """
 
     @task(outlets=[combined_player_stats])
@@ -101,7 +101,7 @@ def combine_player_stats(dag_run=None):
 @asset(
     uri="file://analytics/player-stats/computed-player-odds.csv",
     # Fallback to IdentityMapper if no partition_mapper is specified.
-    # If we want to other temporal mapper (e.g., ToHourlyMapper) here,
+    # If we want to other temporal mapper (e.g., StartOfHourMapper) here,
     # make sure the input_format is changed since the partition_key is now in "%Y-%m-%dT%H" format
     # instead of a valid timestamp
     schedule=PartitionedAssetTimetable(assets=combined_player_stats),
@@ -121,9 +121,9 @@ def compute_player_odds():
     schedule=PartitionedAssetTimetable(
         assets=(combined_player_stats & team_a_player_stats & Asset.ref(name="team_b_player_stats")),
         partition_mapper_config={
-            combined_player_stats: ToYearlyMapper(),  # incompatible on purpose
-            team_a_player_stats: ToHourlyMapper(),
-            Asset.ref(name="team_b_player_stats"): ToHourlyMapper(),
+            combined_player_stats: StartOfYearMapper(),  # incompatible on purpose
+            team_a_player_stats: StartOfHourMapper(),
+            Asset.ref(name="team_b_player_stats"): StartOfHourMapper(),
         },
     ),
     catchup=False,
@@ -164,7 +164,7 @@ def ingest_sales():
     dag_id="aggregate_regional_sales",
     schedule=PartitionedAssetTimetable(
         assets=regional_sales,
-        default_partition_mapper=ProductMapper(IdentityMapper(), ToDailyMapper()),
+        default_partition_mapper=ProductMapper(IdentityMapper(), StartOfDayMapper()),
     ),
     catchup=False,
     tags=["sales", "aggregation"],
@@ -173,7 +173,7 @@ def ingest_sales():
     Aggregate regional sales using ProductMapper.
 
     The ProductMapper splits the composite key "region|timestamp" and applies
-    IdentityMapper to the region segment and ToDailyMapper to the timestamp segment,
+    IdentityMapper to the region segment and StartOfDayMapper to the timestamp segment,
     aligning hourly partitions to daily granularity per region.
     """
 

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

@@ -63,7 +63,7 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
         )
 
 
-class ToHourlyMapper(_BaseTemporalMapper):
+class StartOfHourMapper(_BaseTemporalMapper):
     """Map a time-based partition key to hour."""
 
     default_output_format = "%Y-%m-%dT%H"
@@ -72,7 +72,7 @@ def normalize(self, dt: datetime) -> datetime:
         return dt.replace(minute=0, second=0, microsecond=0)
 
 
-class ToDailyMapper(_BaseTemporalMapper):
+class StartOfDayMapper(_BaseTemporalMapper):
     """Map a time-based partition key to day."""
 
     default_output_format = "%Y-%m-%d"
@@ -81,7 +81,7 @@ def normalize(self, dt: datetime) -> datetime:
         return dt.replace(hour=0, minute=0, second=0, microsecond=0)
 
 
-class ToWeeklyMapper(_BaseTemporalMapper):
+class StartOfWeekMapper(_BaseTemporalMapper):
     """Map a time-based partition key to week."""
 
     default_output_format = "%Y-%m-%d (W%V)"
@@ -91,7 +91,7 @@ def normalize(self, dt: datetime) -> datetime:
         return start.replace(hour=0, minute=0, second=0, microsecond=0)
 
 
-class ToMonthlyMapper(_BaseTemporalMapper):
+class StartOfMonthMapper(_BaseTemporalMapper):
     """Map a time-based partition key to month."""
 
     default_output_format = "%Y-%m"
@@ -106,7 +106,7 @@ def normalize(self, dt: datetime) -> datetime:
         )
 
 
-class ToQuarterlyMapper(_BaseTemporalMapper):
+class StartOfQuarterMapper(_BaseTemporalMapper):
     """Map a time-based partition key to quarter."""
 
     default_output_format = "%Y-Q{quarter}"
@@ -128,7 +128,7 @@ def format(self, dt: datetime) -> str:
         return dt.strftime(self.output_format).format(quarter=quarter)
 
 
-class ToYearlyMapper(_BaseTemporalMapper):
+class StartOfYearMapper(_BaseTemporalMapper):
     """Map a time-based partition key to year."""
 
     default_output_format = "%Y"

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

@@ -44,15 +44,15 @@
     MultipleCronTriggerTimetable,
     PartitionMapper,
     ProductMapper,
-    ToDailyMapper,
-    ToMonthlyMapper,
-    ToQuarterlyMapper,
-    ToWeeklyMapper,
-    ToYearlyMapper,
+    StartOfDayMapper,
+    StartOfMonthMapper,
+    StartOfQuarterMapper,
+    StartOfWeekMapper,
+    StartOfYearMapper,
 )
 from airflow.sdk.bases.timetable import BaseTimetable
 from airflow.sdk.definitions.asset import AssetRef
-from airflow.sdk.definitions.partition_mappers.temporal import ToHourlyMapper
+from airflow.sdk.definitions.partition_mappers.temporal import StartOfHourMapper
 from airflow.sdk.definitions.timetables.assets import (
     AssetTriggeredTimetable,
     PartitionedAssetTimetable,
@@ -393,16 +393,16 @@ 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",
         IdentityMapper: "airflow.partition_mappers.identity.IdentityMapper",
-        ToHourlyMapper: "airflow.partition_mappers.temporal.ToHourlyMapper",
-        ToDailyMapper: "airflow.partition_mappers.temporal.ToDailyMapper",
-        ToWeeklyMapper: "airflow.partition_mappers.temporal.ToWeeklyMapper",
-        ToMonthlyMapper: "airflow.partition_mappers.temporal.ToMonthlyMapper",
-        ToQuarterlyMapper: "airflow.partition_mappers.temporal.ToQuarterlyMapper",
-        ToYearlyMapper: "airflow.partition_mappers.temporal.ToYearlyMapper",
         ProductMapper: "airflow.partition_mappers.product.ProductMapper",
-        AllowedKeyMapper: "airflow.partition_mappers.allowed_key.AllowedKeyMapper",
+        StartOfDayMapper: "airflow.partition_mappers.temporal.StartOfDayMapper",
+        StartOfHourMapper: "airflow.partition_mappers.temporal.StartOfHourMapper",
+        StartOfMonthMapper: "airflow.partition_mappers.temporal.StartOfMonthMapper",
+        StartOfQuarterMapper: "airflow.partition_mappers.temporal.StartOfQuarterMapper",
+        StartOfWeekMapper: "airflow.partition_mappers.temporal.StartOfWeekMapper",
+        StartOfYearMapper: "airflow.partition_mappers.temporal.StartOfYearMapper",
     }
 
     @functools.singledispatchmethod
@@ -421,20 +421,20 @@ def _(self, partition_mapper: ChainMapper) -> dict[str, Any]:
     def _(self, partition_mapper: IdentityMapper) -> dict[str, Any]:
         return {}
 
-    @serialize_partition_mapper.register(ToHourlyMapper)
-    @serialize_partition_mapper.register(ToDailyMapper)
-    @serialize_partition_mapper.register(ToWeeklyMapper)
-    @serialize_partition_mapper.register(ToMonthlyMapper)
-    @serialize_partition_mapper.register(ToQuarterlyMapper)
-    @serialize_partition_mapper.register(ToYearlyMapper)
+    @serialize_partition_mapper.register(StartOfHourMapper)
+    @serialize_partition_mapper.register(StartOfDayMapper)
+    @serialize_partition_mapper.register(StartOfWeekMapper)
+    @serialize_partition_mapper.register(StartOfMonthMapper)
+    @serialize_partition_mapper.register(StartOfQuarterMapper)
+    @serialize_partition_mapper.register(StartOfYearMapper)
     def _(
         self,
-        partition_mapper: ToHourlyMapper
-        | ToDailyMapper
-        | ToWeeklyMapper
-        | ToMonthlyMapper
-        | ToQuarterlyMapper
-        | ToYearlyMapper,
+        partition_mapper: StartOfHourMapper
+        | StartOfDayMapper
+        | StartOfWeekMapper
+        | StartOfMonthMapper
+        | StartOfQuarterMapper
+        | StartOfYearMapper,
     ) -> dict[str, Any]:
         return {
             "input_format": partition_mapper.input_format,

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -92,7 +92,7 @@
     AssetWatcher,
     CronPartitionTimetable,
     IdentityMapper,
-    ToHourlyMapper,
+    StartOfHourMapper,
     task,
 )
 from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
@@ -8982,7 +8982,7 @@ def test_partitioned_dag_run_with_invalid_mapping(dag_maker: DagMaker, session:
         dag_id="asset-event-consumer",
         schedule=PartitionedAssetTimetable(
             assets=asset_1,
-            default_partition_mapper=ToHourlyMapper(),
+            default_partition_mapper=StartOfHourMapper(),
         ),
         session=session,
     ):

airflow-core/tests/unit/partition_mappers/test_chain.py

@@ -22,7 +22,7 @@
 from airflow.partition_mappers.base import PartitionMapper
 from airflow.partition_mappers.chain import ChainMapper
 from airflow.partition_mappers.identity import IdentityMapper
-from airflow.partition_mappers.temporal import ToDailyMapper, ToHourlyMapper
+from airflow.partition_mappers.temporal import StartOfDayMapper, StartOfHourMapper
 
 
 class _InvalidReturnMapper(PartitionMapper):
@@ -37,7 +37,7 @@ def to_downstream(self, key: str) -> list[None]:  # type: ignore[override]
 
 class TestChainMapper:
     def test_to_downstream(self):
-        sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
+        sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
         assert sm.to_downstream("2024-01-15T10:30:00") == "2024-01-15"
 
     def test_to_downstream_invalid_non_iterable_return(self):
@@ -53,17 +53,17 @@ def test_to_downstream_invalid_iterable_contents(self):
     def test_serialize(self):
         from airflow.serialization.encoders import encode_partition_mapper
 
-        sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
+        sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
         result = sm.serialize()
         assert result == {
             "mappers": [
-                encode_partition_mapper(ToHourlyMapper()),
-                encode_partition_mapper(ToDailyMapper(input_format="%Y-%m-%dT%H")),
+                encode_partition_mapper(StartOfHourMapper()),
+                encode_partition_mapper(StartOfDayMapper(input_format="%Y-%m-%dT%H")),
             ],
         }
 
     def test_deserialize(self):
-        sm = ChainMapper(ToHourlyMapper(), ToDailyMapper(input_format="%Y-%m-%dT%H"))
+        sm = ChainMapper(StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"))
         serialized = sm.serialize()
         restored = ChainMapper.deserialize(serialized)
         assert isinstance(restored, ChainMapper)