Add feature views

protos/feast/core/FeatureView.proto

@@ -0,0 +1,70 @@
+//
+// Copyright 2020 The Feast Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+
+syntax = "proto3";
+package feast.core;
+
+option go_package = "github.com/feast-dev/feast/sdk/go/protos/feast/core";
+option java_outer_classname = "FeatureViewProto";
+option java_package = "feast.proto.core";
+
+import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
+import "feast/core/DataSource.proto";
+import "feast/core/Feature.proto";
+
+message FeatureView {
+    // User-specified specifications of this feature table.
+    FeatureViewSpec spec = 1;
+
+    // System-populated metadata for this feature table.
+    FeatureViewMeta meta = 2;
+    string project = 3;
+}
+
+message FeatureViewSpec {
+    // Name of the feature table. Must be unique. Not updated.
+    string name = 1;
+
+    // List names of entities to associate with the Features defined in this 
+    // Feature View. Not updatable.
+    repeated string entities = 3;
+
+    // List of features specifications for each feature defined with this feature table.
+    repeated FeatureSpecV2 features = 4;
+
+    // User defined metadata
+    map<string,string> tags = 5;
+
+    // Features in this feature table can only be retrieved from online serving
+    // younger than ttl. Ttl is measured as the duration of time between
+    // the feature's event timestamp and when the feature is retrieved
+    // Feature values outside ttl will be returned as unset values and indicated to end user
+    google.protobuf.Duration ttl = 6;
+
+    DataSource input = 7;
+
+    bool online = 8;
+}
+
+message FeatureViewMeta {
+    // Time where this Feature View is created
+    google.protobuf.Timestamp created_timestamp = 1;
+
+    // Time where this Feature View is last updated
+    google.protobuf.Timestamp last_updated_timestamp = 2;
+}

protos/feast/core/Registry.proto

@@ -23,11 +23,13 @@ option go_package = "github.com/feast-dev/feast/sdk/go/protos/feast/core";
 
 import "feast/core/Entity.proto";
 import "feast/core/FeatureTable.proto";
+import "feast/core/FeatureView.proto";
 import "google/protobuf/timestamp.proto";
 
 message Registry {
     repeated Entity entities = 1;
     repeated FeatureTable feature_tables = 2;
+    repeated FeatureView feature_views = 6;
 
     string registry_schema_version = 3; // to support migrations; incremented when schema is changed
     string version_id = 4; // version id, random string generated on each update of the data; now used only for debugging purposes

sdk/python/feast/__init__.py

@@ -11,6 +11,7 @@
 from .entity import Entity
 from .feature import Feature
 from .feature_table import FeatureTable
+from .feature_view import FeatureView
 from .value_type import ValueType
 
 try:
@@ -28,6 +29,7 @@
     "KinesisSource",
     "Feature",
     "FeatureTable",
+    "FeatureView",
     "SourceType",
     "ValueType",
 ]

sdk/python/feast/data_source.py

@@ -579,6 +579,10 @@ def __eq__(self, other):
 
         return True
 
+    @property
+    def table_ref(self):
+        return self._bigquery_options.table_ref
+
     @property
     def bigquery_options(self):
         """

sdk/python/feast/feature_view.py

@@ -11,43 +11,138 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from datetime import datetime
-from typing import Dict, List
+from datetime import timedelta
+from typing import Dict, List, Optional, Union
 
-from feast.big_query_source import BigQuerySource
-from feast.entity import Entity
+from google.protobuf.duration_pb2 import Duration
+from google.protobuf.timestamp_pb2 import Timestamp
+
+from feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
+from feast.core.FeatureView_pb2 import FeatureViewMeta as FeatureViewMetaProto
+from feast.core.FeatureView_pb2 import FeatureViewSpec as FeatureViewSpecProto
+from feast.data_source import BigQuerySource, DataSource
 from feast.feature import Feature
+from feast.value_type import ValueType
 
 
 class FeatureView:
     """
     A FeatureView defines a logical grouping of servable features.
     """
 
+    name: str
+    entities: List[str]
+    features: List[Feature]
+    tags: Dict[str, str]
+    ttl: Optional[Duration]
+    online: bool
+    input: BigQuerySource
+
+    created_timestamp: Optional[Timestamp] = None
+    last_updated_timestamp: Optional[Timestamp] = None
+
     def __init__(
         self,
         name: str,
-        entities: List[Entity],
+        entities: List[str],
         features: List[Feature],
         tags: Dict[str, str],
-        ttl: str,
+        ttl: Optional[Union[Duration, timedelta]],
         online: bool,
-        inputs: BigQuerySource,
-        feature_start_time: datetime,
+        input: BigQuerySource,
     ):
-        cols = [entity.name for entity in entities] + [feat.name for feat in features]
+        cols = [entity for entity in entities] + [feat.name for feat in features]
         for col in cols:
-            if inputs.field_mapping is not None and col in inputs.field_mapping.keys():
+            if input.field_mapping is not None and col in input.field_mapping.keys():
                 raise ValueError(
-                    f"The field {col} is mapped to {inputs.field_mapping[col]} for this data source. Please either remove this field mapping or use {inputs.field_mapping[col]} as the Entity or Feature name."
+                    f"The field {col} is mapped to {input.field_mapping[col]} for this data source. Please either remove this field mapping or use {input.field_mapping[col]} as the Entity or Feature name."
                 )
 
         self.name = name
         self.entities = entities
         self.features = features
         self.tags = tags
-        self.ttl = ttl
+        if isinstance(ttl, timedelta):
+            proto_ttl = Duration()
+            proto_ttl.FromTimedelta(ttl)
+            self.ttl = proto_ttl
+        else:
+            self.ttl = ttl
+
         self.online = online
-        self.inputs = inputs
-        self.feature_start_time = feature_start_time
-        return
+        self.input = input
+
+    def is_valid(self):
+        """
+        Validates the state of a feature view locally. Raises an exception
+        if feature view is invalid.
+        """
+
+        if not self.name:
+            raise ValueError("Feature view needs a name")
+
+        if not self.entities:
+            raise ValueError("Feature view has no entities")
+
+    def to_proto(self) -> FeatureViewProto:
+        """
+        Converts an feature view object to its protobuf representation.
+
+        Returns:
+            FeatureViewProto protobuf
+        """
+
+        meta = FeatureViewMetaProto(
+            created_timestamp=self.created_timestamp,
+            last_updated_timestamp=self.last_updated_timestamp,
+        )
+
+        spec = FeatureViewSpecProto(
+            name=self.name,
+            entities=self.entities,
+            features=[feature.to_proto() for feature in self.features],
+            tags=self.tags,
+            ttl=self.ttl,
+            online=self.online,
+            input=self.input.to_proto(),
+        )
+
+        return FeatureViewProto(spec=spec, meta=meta)
+
+    @classmethod
+    def from_proto(cls, feature_view_proto: FeatureViewProto):
+        """
+        Creates a feature view from a protobuf representation of a feature view
+
+        Args:
+            feature_view_proto: A protobuf representation of a feature view
+
+        Returns:
+            Returns a FeatureViewProto object based on the feature view protobuf
+        """
+
+        feature_view = cls(
+            name=feature_view_proto.spec.name,
+            entities=[entity for entity in feature_view_proto.spec.entities],
+            features=[
+                Feature(
+                    name=feature.name,
+                    dtype=ValueType(feature.value_type),
+                    labels=feature.labels,
+                )
+                for feature in feature_view_proto.spec.features
+            ],
+            tags=dict(feature_view_proto.spec.tags),
+            online=feature_view_proto.spec.online,
+            ttl=(
+                None
+                if feature_view_proto.spec.ttl.seconds == 0
+                and feature_view_proto.spec.ttl.nanos == 0
+                else feature_view_proto.spec.ttl
+            ),
+            input=DataSource.from_proto(feature_view_proto.spec.input),
+        )
+
+        feature_view.created_timestamp = feature_view_proto.meta.created_timestamp
+
+        return feature_view

sdk/python/feast/infra/gcp.py

@@ -1,10 +1,10 @@
 from datetime import datetime
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
 
 import mmh3
 from pytz import utc
 
-from feast import FeatureTable
+from feast import FeatureTable, FeatureView
 from feast.infra.provider import Provider
 from feast.repo_config import DatastoreOnlineStoreConfig
 from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
@@ -66,8 +66,8 @@ def _initialize_client(self):
     def update_infra(
         self,
         project: str,
-        tables_to_delete: List[FeatureTable],
-        tables_to_keep: List[FeatureTable],
+        tables_to_delete: List[Union[FeatureTable, FeatureView]],
+        tables_to_keep: List[Union[FeatureTable, FeatureView]],
     ):
         from google.cloud import datastore
 
@@ -88,7 +88,9 @@ def update_infra(
             key = client.key("Project", project, "Table", table.name)
             client.delete(key)
 
-    def teardown_infra(self, project: str, tables: List[FeatureTable]) -> None:
+    def teardown_infra(
+        self, project: str, tables: List[Union[FeatureTable, FeatureView]]
+    ) -> None:
         client = self._initialize_client()
 
         for table in tables:
@@ -103,7 +105,7 @@ def teardown_infra(self, project: str, tables: List[FeatureTable]) -> None:
     def online_write_batch(
         self,
         project: str,
-        table: FeatureTable,
+        table: Union[FeatureTable, FeatureView],
         data: List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]],
         created_ts: datetime,
     ) -> None:
@@ -143,7 +145,10 @@ def online_write_batch(
                 client.put(entity)
 
     def online_read(
-        self, project: str, table: FeatureTable, entity_key: EntityKeyProto
+        self,
+        project: str,
+        table: Union[FeatureTable, FeatureView],
+        entity_key: EntityKeyProto,
     ) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:
         client = self._initialize_client()