feast-dev · nquinn408
@@ -23,6 +23,8 @@ Feature views consist of:
 * a name to uniquely identify this feature view in the project.
 * (optional, but recommended) a schema specifying one or more [features](feature-view.md#field) (without this, Feast will infer the schema by reading from the data source)
 * (optional, but recommended) metadata (for example, description, or other free-form metadata via `tags`)
+* (optional) `owner`: the email of the primary maintainer
+* (optional) `org`: the organizational unit that owns the feature view (e.g. `"ads"`, `"search"`); useful for grouping feature views by team or product area
 * (optional) a TTL, which limits how far back Feast will look when generating historical datasets
 * (optional) `enable_validation=True`, which enables schema validation during materialization (see [Schema Validation](#schema-validation) below)
 
@@ -270,7 +272,7 @@ def transformed_conv_rate(features_df: pd.DataFrame) -> pd.DataFrame:
 
 A stream feature view is an extension of a normal feature view. The primary difference is that stream feature views have both stream and batch data sources, whereas a normal feature view only has a batch data source.
 
-Stream feature views should be used instead of normal feature views when there are stream data sources (e.g. Kafka and Kinesis) available to provide fresh features in an online setting. Here is an example definition of a stream feature view with an attached transformation:
+Stream feature views should be used instead of normal feature views when there are stream data sources (e.g. Kafka and Kinesis) available to provide fresh features in an online setting. Like regular feature views, stream feature views support the optional `org` field for grouping by organizational unit. Here is an example definition of a stream feature view with an attached transformation:
 
 ```python
 from datetime import timedelta
@@ -308,6 +310,7 @@ driver_stats_stream_source = KafkaSource(
     timestamp_field="event_timestamp",
     online=True,
     source=driver_stats_stream_source,
+    org="ads",  # optional
 )
 def driver_hourly_stats_stream(df: DataFrame):
     from pyspark.sql.functions import col

@@ -36,7 +36,7 @@ message FeatureView {
     FeatureViewMeta meta = 2;
 }
 
-// Next available id: 19
+// Next available id: 20
 // TODO(adchia): refactor common fields from this and ODFV into separate metadata proto
 message FeatureViewSpec {
     // Name of the feature view. Must be unique. Not updated.
@@ -97,6 +97,9 @@ message FeatureViewSpec {
 
     // User-specified version pin (e.g. "latest", "v2", "version2")
     string version = 18;
+
+    // Organizational unit that owns this feature view (e.g. "ads", "search").
+    string org = 19;
 }
 
 message FeatureViewMeta {

@@ -36,7 +36,7 @@ message OnDemandFeatureView {
     OnDemandFeatureViewMeta meta = 2;
 }
 
-// Next available id: 18
+// Next available id: 19
 message OnDemandFeatureViewSpec {
     // Name of the feature view. Must be unique. Not updated.
     string name = 1;
@@ -77,6 +77,9 @@ message OnDemandFeatureViewSpec {
 
     // User-specified version pin (e.g. "latest", "v2", "version2")
     string version = 17;
+
+    // Organizational unit that owns this feature view (e.g. "ads", "search").
+    string org = 18;
 }
 
 message OnDemandFeatureViewMeta {

@@ -37,7 +37,7 @@ message StreamFeatureView {
     FeatureViewMeta meta = 2;
 }
 
-// Next available id: 22
+// Next available id: 23
 message StreamFeatureViewSpec {
     // Name of the feature view. Must be unique. Not updated.
     string name = 1;
@@ -105,5 +105,8 @@ message StreamFeatureViewSpec {
 
     // User-specified version pin (e.g. "latest", "v2", "version2")
     string version = 21;
+
+    // Organizational unit that owns this stream feature view (e.g. "ads", "search").
+    string org = 22;
 }
 
@@ -93,6 +93,7 @@ def __init__(
         offline: bool = False,
         description: str = "",
         owner: str = "",
+        org: str = "",
         schema: Optional[List[Field]] = None,
         udf: Optional[Callable[[Any], Any]] = None,
         udf_string: Optional[str] = "",
@@ -151,6 +152,7 @@ def __init__(
             offline=offline,
             description=description,
             owner=owner,
+            org=org,
             schema=schema,
             source=source,  # type: ignore[arg-type]
             sink_source=sink_source,
@@ -189,6 +191,7 @@ def batch_feature_view(
     offline: bool = True,
     description: str = "",
     owner: str = "",
+    org: str = "",
     schema: Optional[List[Field]] = None,
     enable_validation: bool = False,
     version: str = "latest",
@@ -219,6 +222,7 @@ def decorator(user_function):
             offline=offline,
             description=description,
             owner=owner,
+            org=org,
             schema=schema,
             udf=user_function,
             udf_string=udf_string,

@@ -89,6 +89,8 @@ class FeatureView(BaseFeatureView):
         tags: A dictionary of key-value pairs to store arbitrary metadata.
         owner: The owner of the feature view, typically the email of the primary
             maintainer.
+        org: The organizational unit that owns this feature view (e.g. "ads", "search").
+            Defaults to empty string.
         mode: The transformation mode for feature transformations. Only meaningful when
             transformations are applied. Choose from TransformationMode enum values
             (e.g., PYTHON, PANDAS, RAY, SQL, SPARK, SUBSTRAIT).
@@ -107,6 +109,7 @@ class FeatureView(BaseFeatureView):
     description: str
     tags: Dict[str, str]
     owner: str
+    org: str
     materialization_intervals: List[Tuple[datetime, datetime]]
     mode: Optional[Union["TransformationMode", str]]
     enable_validation: bool
@@ -125,6 +128,7 @@ def __init__(
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         owner: str = "",
+        org: str = "",
         mode: Optional[Union["TransformationMode", str]] = None,
         enable_validation: bool = False,
         version: str = "latest",
@@ -152,6 +156,8 @@ def __init__(
             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
             owner (optional): The owner of the feature view, typically the email of the
                 primary maintainer.
+            org (optional): The organizational unit that owns this feature view
+                (e.g. "ads", "search").
             mode (optional): The transformation mode for feature transformations. Only meaningful
                 when transformations are applied. Choose from TransformationMode enum values.
             enable_validation (optional): If True, enables schema validation during materialization
@@ -278,6 +284,7 @@ def __init__(
             owner=owner,
             source=self.batch_source,
         )
+        self.org = org
         self.online = online
         self.offline = offline
         self.mode = mode
@@ -302,6 +309,7 @@ def __copy__(self):
             version=self.version,
             description=self.description,
             owner=self.owner,
+            org=self.org,
         )
 
         # This is deliberately set outside of the FV initialization as we do not have the Entity objects.
@@ -355,6 +363,7 @@ def __eq__(self, other):
             or self.enable_validation != other.enable_validation
             or normalize_version_string(self.version)
             != normalize_version_string(other.version)
+            or self.org != other.org
         ):
             return False
 
@@ -485,6 +494,7 @@ def to_proto_spec(
             description=self.description,
             tags=self.tags,
             owner=self.owner,
+            org=self.org,
             ttl=(ttl_duration if ttl_duration is not None else None),
             online=self.online,
             offline=self.offline,
@@ -615,6 +625,7 @@ def _from_proto_internal(
                 description=feature_view_proto.spec.description,
                 tags=dict(feature_view_proto.spec.tags),
                 owner=feature_view_proto.spec.owner,
+                org=feature_view_proto.spec.org,
                 online=feature_view_proto.spec.online,
                 offline=feature_view_proto.spec.offline,
                 ttl=(
@@ -637,6 +648,7 @@ def _from_proto_internal(
                 description=feature_view_proto.spec.description,
                 tags=dict(feature_view_proto.spec.tags),
                 owner=feature_view_proto.spec.owner,
+                org=feature_view_proto.spec.org,
                 online=feature_view_proto.spec.online,
                 offline=feature_view_proto.spec.offline,
                 ttl=(

@@ -131,6 +131,8 @@ class OnDemandFeatureView(BaseFeatureView):
         tags: A dictionary of key-value pairs to store arbitrary metadata.
         owner: The owner of the on demand feature view, typically the email of the primary
             maintainer.
+        org: The organizational unit that owns this on demand feature view (e.g. "ads",
+            "search"). Defaults to empty string.
     """
 
     _TRACK_METRICS_TAG = "feast:track_metrics"
@@ -146,6 +148,7 @@ class OnDemandFeatureView(BaseFeatureView):
     description: str
     tags: dict[str, str]
     owner: str
+    org: str
     write_to_online_store: bool
     singleton: bool
     track_metrics: bool
@@ -168,6 +171,7 @@ def __init__(  # noqa: C901
         description: str = "",
         tags: Optional[dict[str, str]] = None,
         owner: str = "",
+        org: str = "",
         write_to_online_store: bool = False,
         singleton: bool = False,
         track_metrics: bool = False,
@@ -199,6 +203,8 @@ def __init__(  # noqa: C901
             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
             owner (optional): The owner of the on demand feature view, typically the email
                 of the primary maintainer.
+            org (optional): The organizational unit that owns this feature view
+                (e.g. "ads", "search").
             write_to_online_store (optional): A boolean that indicates whether to write the on demand feature view to
             the online store for faster retrieval.
             singleton (optional): A boolean that indicates whether the transformation is executed on a singleton
@@ -218,6 +224,7 @@ def __init__(  # noqa: C901
             owner=owner,
         )
 
+        self.org = org
         self.version = version
         schema = schema or []
         self.entities = [e.name for e in entities] if entities else [DUMMY_ENTITY_NAME]
@@ -384,6 +391,7 @@ def __copy__(self):
             description=self.description,
             tags=self.tags,
             owner=self.owner,
+            org=self.org,
             write_to_online_store=self.write_to_online_store,
             singleton=self.singleton,
             version=self.version,
@@ -463,6 +471,7 @@ def __eq__(self, other):
             or self.aggregations != other.aggregations
             or normalize_version_string(self.version)
             != normalize_version_string(other.version)
+            or self.org != other.org
         ):
             return False
 
@@ -621,6 +630,7 @@ def to_proto(self) -> OnDemandFeatureViewProto:
             description=self.description,
             tags=tags,
             owner=self.owner,
+            org=self.org,
             write_to_online_store=self.write_to_online_store,
             singleton=self.singleton or False,
             aggregations=[agg.to_proto() for agg in self.aggregations],
@@ -689,6 +699,7 @@ def from_proto(
             description=on_demand_feature_view_proto.spec.description,
             tags=proto_tags,
             owner=on_demand_feature_view_proto.spec.owner,
+            org=on_demand_feature_view_proto.spec.org,
             write_to_online_store=optional_fields["write_to_online_store"],
             singleton=optional_fields["singleton"],
             track_metrics=track_metrics,
@@ -1336,6 +1347,7 @@ def on_demand_feature_view(
     description: str = "",
     tags: Optional[dict[str, str]] = None,
     owner: str = "",
+    org: str = "",
     write_to_online_store: bool = False,
     singleton: bool = False,
     track_metrics: bool = False,
@@ -1362,6 +1374,8 @@ def on_demand_feature_view(
         tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
         owner (optional): The owner of the on demand feature view, typically the email
             of the primary maintainer.
+        org (optional): The organizational unit that owns this on demand feature view
+            (e.g. "ads", "search"). Defaults to empty string.
         write_to_online_store (optional): A boolean that indicates whether to write the on demand feature view to
             the online store for faster retrieval.
         singleton (optional): A boolean that indicates whether the transformation is executed on a singleton
@@ -1390,6 +1404,7 @@ def decorator(user_function):
             description=description,
             tags=tags,
             owner=owner,
+            org=org,
             write_to_online_store=write_to_online_store,
             entities=entities,
             singleton=singleton,

@@ -2,44 +2,49 @@
 @generated by mypy-protobuf.  Do not edit manually!
 isort:skip_file
 """
-import builtins
-import google.protobuf.descriptor
-import google.protobuf.duration_pb2
-import google.protobuf.message
+
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import duration_pb2 as _duration_pb2
+from google.protobuf import message as _message
+import builtins as _builtins
 import sys
+import typing as _typing
 
-if sys.version_info >= (3, 8):
-    import typing as typing_extensions
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias as _TypeAlias
 else:
-    import typing_extensions
+    from typing_extensions import TypeAlias as _TypeAlias
 
-DESCRIPTOR: google.protobuf.descriptor.FileDescriptor
+DESCRIPTOR: _descriptor.FileDescriptor
 
-class Aggregation(google.protobuf.message.Message):
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+@_typing.final
+class Aggregation(_message.Message):
+    DESCRIPTOR: _descriptor.Descriptor
 
-    COLUMN_FIELD_NUMBER: builtins.int
-    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
-    @property
-    def time_window(self) -> google.protobuf.duration_pb2.Duration: ...
-    @property
-    def slide_interval(self) -> google.protobuf.duration_pb2.Duration: ...
-    name: builtins.str
+    COLUMN_FIELD_NUMBER: _builtins.int
+    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
+    @_builtins.property
+    def time_window(self) -> _duration_pb2.Duration: ...
+    @_builtins.property
+    def slide_interval(self) -> _duration_pb2.Duration: ...
     def __init__(
         self,
         *,
-        column: builtins.str = ...,
-        function: builtins.str = ...,
-        time_window: google.protobuf.duration_pb2.Duration | None = ...,
-        slide_interval: google.protobuf.duration_pb2.Duration | None = ...,
-        name: builtins.str = ...,
+        column: _builtins.str = ...,
+        function: _builtins.str = ...,
+        time_window: _duration_pb2.Duration | None = ...,
+        slide_interval: _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", "name", b"name", "slide_interval", b"slide_interval", "time_window", b"time_window"]) -> None: ...
+    _HasFieldArgType: _TypeAlias = _typing.Literal["slide_interval", b"slide_interval", "time_window", b"time_window"]  # noqa: Y015
+    def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ...
+    _ClearFieldArgType: _TypeAlias = _typing.Literal["column", b"column", "function", b"function", "name", b"name", "slide_interval", b"slide_interval", "time_window", b"time_window"]  # noqa: Y015
+    def ClearField(self, field_name: _ClearFieldArgType) -> None: ...
 
-global___Aggregation = Aggregation
+Global___Aggregation: _TypeAlias = Aggregation  # noqa: Y015