feat: Add aggregation in OnDemandFeatureView

protos/feast/core/OnDemandFeatureView.proto

@@ -28,6 +28,7 @@ import "feast/core/FeatureViewProjection.proto";
 import "feast/core/Feature.proto";
 import "feast/core/DataSource.proto";
 import "feast/core/Transformation.proto";
+import "feast/core/Aggregation.proto";
 
 message OnDemandFeatureView {
     // User-specified specifications of this feature view.
@@ -70,6 +71,10 @@ message OnDemandFeatureViewSpec {
     // List of specifications for each entity defined as part of this feature view.
     repeated FeatureSpecV2 entity_columns = 14;
     bool singleton = 15;
+
+    // Aggregation definitions
+    repeated Aggregation aggregations = 16;
+
 }
 
 message OnDemandFeatureViewMeta {

sdk/python/feast/on_demand_feature_view.py

@@ -8,6 +8,7 @@
 import pyarrow
 from typeguard import typechecked
 
+from feast.aggregation import Aggregation
 from feast.base_feature_view import BaseFeatureView
 from feast.data_source import RequestSource
 from feast.entity import Entity
@@ -76,6 +77,7 @@ class OnDemandFeatureView(BaseFeatureView):
     singleton: bool
     udf: Optional[FunctionType]
     udf_string: Optional[str]
+    aggregations: List[Aggregation]
 
     def __init__(  # noqa: C901
         self,
@@ -93,6 +95,7 @@ def __init__(  # noqa: C901
         owner: str = "",
         write_to_online_store: bool = False,
         singleton: bool = False,
+        aggregations: Optional[List[Aggregation]] = None,
     ):
         """
         Creates an OnDemandFeatureView object.
@@ -118,6 +121,7 @@ def __init__(  # noqa: C901
             the online store for faster retrieval.
             singleton (optional): A boolean that indicates whether the transformation is executed on a singleton
                 (only applicable when mode="python").
+            aggregations (optional): List of aggregations to apply before transformation.
         """
         super().__init__(
             name=name,
@@ -187,6 +191,7 @@ def __init__(  # noqa: C901
         self.singleton = singleton
         if self.singleton and self.mode != "python":
             raise ValueError("Singleton is only supported for Python mode.")
+        self.aggregations = aggregations or []
 
     def get_feature_transformation(self) -> Transformation:
         if not self.udf:
@@ -251,6 +256,7 @@ def __eq__(self, other):
             or self.write_to_online_store != other.write_to_online_store
             or sorted(self.entity_columns) != sorted(other.entity_columns)
             or self.singleton != other.singleton
+            or self.aggregations != other.aggregations
         ):
             return False
 
@@ -342,6 +348,7 @@ def to_proto(self) -> OnDemandFeatureViewProto:
             owner=self.owner,
             write_to_online_store=self.write_to_online_store,
             singleton=self.singleton if self.singleton else False,
+            aggregations=self.aggregations,
         )
         return OnDemandFeatureViewProto(spec=spec, meta=meta)
 
@@ -451,6 +458,12 @@ def from_proto(
         if hasattr(on_demand_feature_view_proto.spec, "singleton"):
             singleton = on_demand_feature_view_proto.spec.singleton
 
+        aggregations = []
+        if hasattr(on_demand_feature_view_proto.spec, "aggregations"):
+            aggregations = [
+                Aggregation.from_proto(aggregation_proto)
+                for aggregation_proto in on_demand_feature_view_proto.spec.aggregations
+            ]
         on_demand_feature_view_obj = cls(
             name=on_demand_feature_view_proto.spec.name,
             schema=[
@@ -471,6 +484,7 @@ def from_proto(
             owner=on_demand_feature_view_proto.spec.owner,
             write_to_online_store=write_to_online_store,
             singleton=singleton,
+            aggregations=aggregations,
         )
 
         on_demand_feature_view_obj.entities = entities

sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi

@@ -71,7 +71,7 @@ class FeatureServiceSpec(google.protobuf.message.Message):
     """Name of Feast project that this Feature Service belongs to."""
     @property
     def features(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.core.FeatureViewProjection_pb2.FeatureViewProjection]:
-        """Represents a projection that's to be applied on top of the FeatureView.
+        """Represents a projection that's to be applied on top of the FeatureView.
         Contains data such as the features to use from a FeatureView.
         """
     @property

sdk/python/feast/protos/feast/core/OnDemandFeatureView_pb2.pyi

@@ -18,6 +18,7 @@ limitations under the License.
 """
 import builtins
 import collections.abc
+import feast.core.Aggregation_pb2
 import feast.core.DataSource_pb2
 import feast.core.FeatureViewProjection_pb2
 import feast.core.FeatureView_pb2
@@ -108,6 +109,7 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
     ENTITIES_FIELD_NUMBER: builtins.int
     ENTITY_COLUMNS_FIELD_NUMBER: builtins.int
     SINGLETON_FIELD_NUMBER: builtins.int
+    AGGREGATIONS_FIELD_NUMBER: builtins.int
     name: builtins.str
     """Name of the feature view. Must be unique. Not updated."""
     project: builtins.str
@@ -139,6 +141,9 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
     def entity_columns(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.core.Feature_pb2.FeatureSpecV2]:
         """List of specifications for each entity defined as part of this feature view."""
     singleton: builtins.bool
+    @property
+    def aggregations(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.core.Aggregation_pb2.Aggregation]:
+        """Aggregation definitions"""
     def __init__(
         self,
         *,
@@ -156,9 +161,10 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
         entities: collections.abc.Iterable[builtins.str] | None = ...,
         entity_columns: collections.abc.Iterable[feast.core.Feature_pb2.FeatureSpecV2] | None = ...,
         singleton: builtins.bool = ...,
+        aggregations: collections.abc.Iterable[feast.core.Aggregation_pb2.Aggregation] | None = ...,
     ) -> None: ...
     def HasField(self, field_name: typing_extensions.Literal["feature_transformation", b"feature_transformation", "user_defined_function", b"user_defined_function"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "entities", b"entities", "entity_columns", b"entity_columns", "feature_transformation", b"feature_transformation", "features", b"features", "mode", b"mode", "name", b"name", "owner", b"owner", "project", b"project", "singleton", b"singleton", "sources", b"sources", "tags", b"tags", "user_defined_function", b"user_defined_function", "write_to_online_store", b"write_to_online_store"]) -> None: ...
+    def ClearField(self, field_name: typing_extensions.Literal["aggregations", b"aggregations", "description", b"description", "entities", b"entities", "entity_columns", b"entity_columns", "feature_transformation", b"feature_transformation", "features", b"features", "mode", b"mode", "name", b"name", "owner", b"owner", "project", b"project", "singleton", b"singleton", "sources", b"sources", "tags", b"tags", "user_defined_function", b"user_defined_function", "write_to_online_store", b"write_to_online_store"]) -> None: ...
 
 global___OnDemandFeatureViewSpec = OnDemandFeatureViewSpec