Add streaming sources to the FeatureView API (#1664)

* Add a streaming source to the FeatureView API

This diff only updates the API. It is currently up to the providers to actually use this information to spin up resources to consume events from the stream sources.

Signed-off-by: Achal Shah <achals@gmail.com>

* remove stuff from rebase

Signed-off-by: Achal Shah <achals@gmail.com>

* make format

Signed-off-by: Achal Shah <achals@gmail.com>

* Update protos

Signed-off-by: Achal Shah <achals@gmail.com>

* lint

Signed-off-by: Achal Shah <achals@gmail.com>

* format

Signed-off-by: Achal Shah <achals@gmail.com>

* CR

Signed-off-by: Achal Shah <achals@gmail.com>

* fix test

Signed-off-by: Achal Shah <achals@gmail.com>

* lint

Signed-off-by: Achal Shah <achals@gmail.com>

protos/feast/core/FeatureView.proto

@@ -59,7 +59,9 @@ message FeatureViewSpec {
     google.protobuf.Duration ttl = 6;
 
     // Batch/Offline DataSource where this view can retrieve offline feature data.
-    DataSource input = 7;
+    DataSource batch_source = 7;
+    // Streaming DataSource from where this view can consume "online" feature data.
+    DataSource stream_source = 9;
 
     // Whether these features should be served online or not
     bool online = 8;

sdk/python/feast/data_source.py

@@ -869,6 +869,9 @@ def __init__(
         )
 
     def __eq__(self, other):
+        if other is None:
+            return False
+
         if not isinstance(other, KinesisSource):
             raise TypeError(
                 "Comparisons should only involve KinesisSource class objects."

sdk/python/feast/feature_view.py

@@ -50,7 +50,8 @@ class FeatureView:
     ttl: Optional[timedelta]
     online: bool
     input: DataSource
-
+    batch_source: Optional[DataSource] = None
+    stream_source: Optional[DataSource] = None
     created_timestamp: Optional[Timestamp] = None
     last_updated_timestamp: Optional[Timestamp] = None
     materialization_intervals: List[Tuple[datetime, datetime]]
@@ -62,15 +63,22 @@ def __init__(
         entities: List[str],
         ttl: Optional[Union[Duration, timedelta]],
         input: DataSource,
+        batch_source: Optional[DataSource] = None,
+        stream_source: Optional[DataSource] = None,
         features: List[Feature] = [],
         tags: Optional[Dict[str, str]] = None,
         online: bool = True,
     ):
+        _input = input or batch_source
+        assert _input is not None
+
         cols = [entity for entity in entities] + [feat.name for feat in features]
         for col in cols:
-            if input.field_mapping is not None and col in input.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 {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."
+                    f"The field {col} is mapped to {_input.field_mapping[col]} for this data source. "
+                    f"Please either remove this field mapping or use {_input.field_mapping[col]} as the "
+                    f"Entity or Feature name."
                 )
 
         self.name = name
@@ -84,7 +92,9 @@ def __init__(
             self.ttl = ttl
 
         self.online = online
-        self.input = input
+        self.input = _input
+        self.batch_source = _input
+        self.stream_source = stream_source
 
         self.materialization_intervals = []
 
@@ -118,6 +128,8 @@ def __eq__(self, other):
             return False
         if self.input != other.input:
             return False
+        if self.stream_source != other.stream_source:
+            return False
 
         return True
 
@@ -157,14 +169,21 @@ def to_proto(self) -> FeatureViewProto:
             ttl_duration = Duration()
             ttl_duration.FromTimedelta(self.ttl)
 
+        print(f"Stream soruce: {self.stream_source}, {type(self.stream_source)}")
+
         spec = FeatureViewSpecProto(
             name=self.name,
             entities=self.entities,
             features=[feature.to_proto() for feature in self.features],
             tags=self.tags,
             ttl=(ttl_duration if ttl_duration is not None else None),
             online=self.online,
-            input=self.input.to_proto(),
+            batch_source=self.input.to_proto(),
+            stream_source=(
+                self.stream_source.to_proto()
+                if self.stream_source is not None
+                else None
+            ),
         )
 
         return FeatureViewProto(spec=spec, meta=meta)
@@ -181,6 +200,12 @@ def from_proto(cls, feature_view_proto: FeatureViewProto):
             Returns a FeatureViewProto object based on the feature view protobuf
         """
 
+        _input = DataSource.from_proto(feature_view_proto.spec.batch_source)
+        stream_source = (
+            DataSource.from_proto(feature_view_proto.spec.stream_source)
+            if feature_view_proto.spec.HasField("stream_source")
+            else None
+        )
         feature_view = cls(
             name=feature_view_proto.spec.name,
             entities=[entity for entity in feature_view_proto.spec.entities],
@@ -200,7 +225,9 @@ def from_proto(cls, feature_view_proto: FeatureViewProto):
                 and feature_view_proto.spec.ttl.nanos == 0
                 else feature_view_proto.spec.ttl
             ),
-            input=DataSource.from_proto(feature_view_proto.spec.input),
+            input=_input,
+            batch_source=_input,
+            stream_source=stream_source,
         )
 
         feature_view.created_timestamp = feature_view_proto.meta.created_timestamp