feat: Adding optional name to Aggregation (#5994)

Signed-off-by: Nick Quinn <nicholas_quinn@apple.com>
feast-dev · franciscojavierarceo · Mar 10, 2026 · Mar 9, 2026 · Mar 9, 2026 · Mar 9, 2026
commit 02df4b21f787d0ef6f33c709f17374e7a49a393f
@@ -12,4 +12,5 @@ message Aggregation {
     string function = 2;
     google.protobuf.Duration time_window = 3;
     google.protobuf.Duration slide_interval = 4;
+    string name = 5;
 }
@@ -21,19 +21,22 @@ class Aggregation:
         function: str  # Provided built in aggregations sum, max, min, count mean
         time_window: timedelta  # The time window for this aggregation.
         slide_interval: timedelta # The sliding window for these aggregations
+        name: str  # Optional override for the output feature name (defaults to {function}_{column})
     """
 
     column: str
     function: str
     time_window: Optional[timedelta]
     slide_interval: Optional[timedelta]
+    name: str
 
     def __init__(
         self,
         column: Optional[str] = "",
         function: Optional[str] = "",
         time_window: Optional[timedelta] = None,
         slide_interval: Optional[timedelta] = None,
+        name: Optional[str] = None,
     ):
         self.column = column or ""
         self.function = function or ""
@@ -42,6 +45,7 @@ def __init__(
             self.slide_interval = self.time_window
         else:
             self.slide_interval = slide_interval
+        self.name = name or ""
 
     def to_proto(self) -> AggregationProto:
         window_duration = None
@@ -59,6 +63,7 @@ def to_proto(self) -> AggregationProto:
             function=self.function,
             time_window=window_duration,
             slide_interval=slide_interval_duration,
+            name=self.name,
         )
 
     @classmethod
@@ -79,6 +84,7 @@ def from_proto(cls, agg_proto: AggregationProto):
             function=agg_proto.function,
             time_window=time_window,
             slide_interval=slide_interval,
+            name=agg_proto.name or None,
         )
         return aggregation
 
@@ -91,6 +97,7 @@ def __eq__(self, other):
             or self.function != other.function
             or self.time_window != other.time_window
             or self.slide_interval != other.slide_interval
+            or self.name != other.name
         ):
             return False
 
@@ -106,7 +113,7 @@ def aggregation_specs_to_agg_ops(
     for agg in agg_specs:
         if getattr(agg, "time_window", None) is not None:
             raise ValueError(time_window_unsupported_error_message)
-        alias = f"{agg.function}_{agg.column}"
+        alias = getattr(agg, "name", None) or f"{agg.function}_{agg.column}"
         agg_ops[alias] = (agg.function, agg.column)
     return agg_ops
 

@@ -22,8 +22,10 @@ class Aggregation(google.protobuf.message.Message):
     FUNCTION_FIELD_NUMBER: builtins.int
     TIME_WINDOW_FIELD_NUMBER: builtins.int
     SLIDE_INTERVAL_FIELD_NUMBER: builtins.int
+    NAME_FIELD_NUMBER: builtins.int
     column: builtins.str
     function: builtins.str
+    name: builtins.str
     @property
     def time_window(self) -> google.protobuf.duration_pb2.Duration: ...
     @property
@@ -35,8 +37,9 @@ class Aggregation(google.protobuf.message.Message):
         function: builtins.str = ...,
         time_window: google.protobuf.duration_pb2.Duration | None = ...,
         slide_interval: google.protobuf.duration_pb2.Duration | None = ...,
+        name: builtins.str = ...,
     ) -> None: ...
     def HasField(self, field_name: typing_extensions.Literal["slide_interval", b"slide_interval", "time_window", b"time_window"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["column", b"column", "function", b"function", "slide_interval", b"slide_interval", "time_window", b"time_window"]) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["column", b"column", "function", b"function", "name", b"name", "slide_interval", b"slide_interval", "time_window", b"time_window"]) -> None: ...
 
 global___Aggregation = Aggregation
@@ -1,13 +1,16 @@
+from datetime import timedelta
+
 import pytest
 
-from feast.aggregation import aggregation_specs_to_agg_ops
+from feast.aggregation import Aggregation, aggregation_specs_to_agg_ops
 
 
 class DummyAggregation:
-    def __init__(self, *, function: str, column: str, time_window=None):
+    def __init__(self, *, function: str, column: str, time_window=None, name: str = ""):
         self.function = function
         self.column = column
         self.time_window = time_window
+        self.name = name
 
 
 def test_aggregation_specs_to_agg_ops_success():
@@ -42,3 +45,50 @@ def test_aggregation_specs_to_agg_ops_time_window_unsupported(error_message: str
             agg_specs,
             time_window_unsupported_error_message=error_message,
         )
+
+
+def test_aggregation_specs_to_agg_ops_custom_name():
+    agg_specs = [
+        DummyAggregation(function="sum", column="seconds_watched", name="sum_seconds_watched_per_ad_1d"),
+    ]
+
+    agg_ops = aggregation_specs_to_agg_ops(
+        agg_specs,
+        time_window_unsupported_error_message="Time window aggregation is not supported.",
+    )
+
+    assert agg_ops == {
+        "sum_seconds_watched_per_ad_1d": ("sum", "seconds_watched"),
+    }
+
+
+def test_aggregation_specs_to_agg_ops_mixed_names():
+    agg_specs = [
+        DummyAggregation(function="sum", column="trips", name="total_trips"),
+        DummyAggregation(function="mean", column="fare"),
+    ]
+
+    agg_ops = aggregation_specs_to_agg_ops(
+        agg_specs,
+        time_window_unsupported_error_message="Time window aggregation is not supported.",
+    )
+
+    assert agg_ops == {
+        "total_trips": ("sum", "trips"),
+        "mean_fare": ("mean", "fare"),
+    }
+
+
+def test_aggregation_round_trip_with_name():
+    agg = Aggregation(
+        column="seconds_watched",
+        function="sum",
+        time_window=timedelta(days=1),
+        name="sum_seconds_watched_per_ad_1d",
+    )
+    proto = agg.to_proto()
+    assert proto.name == "sum_seconds_watched_per_ad_1d"
+
+    restored = Aggregation.from_proto(proto)
+    assert restored.name == "sum_seconds_watched_per_ad_1d"
+    assert restored == agg