Skip to content

Commit d024e5e

Browse files
kevjumbakevjumba
authored
fix: Fix bugs in applying stream feature view and retrieving online features (feast-dev#2754)
* Fix apply workflow Signed-off-by: Kevin Zhang <kzhang@tecton.ai> * Fix Signed-off-by: Kevin Zhang <kzhang@tecton.ai> * Fix Signed-off-by: Kevin Zhang <kzhang@tecton.ai> * Fix issues Signed-off-by: Kevin Zhang <kzhang@tecton.ai> * Fix Signed-off-by: Kevin Zhang <kzhang@tecton.ai> * Reformat Signed-off-by: Kevin Zhang <kzhang@tecton.ai>
1 parent 44a3f05 commit d024e5e

File tree

8 files changed

+163
-34
lines changed

8 files changed

+163
-34
lines changed

‎protos/feast/core/StreamFeatureView.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ option java_package = "feast.proto.core";
2626
import "google/protobuf/duration.proto";
2727
import "google/protobuf/timestamp.proto";
2828
import "feast/core/OnDemandFeatureView.proto";
29+
import "feast/core/FeatureView.proto";
2930
import "feast/core/Feature.proto";
3031
import "feast/core/DataSource.proto";
3132
import "feast/core/Aggregation.proto";
@@ -95,4 +96,7 @@ message StreamFeatureViewMeta {
9596

9697
// Time where this Feature View is last updated
9798
google.protobuf.Timestamp last_updated_timestamp = 2;
99+
100+
// List of pairs (start_time, end_time) for which this feature view has been materialized.
101+
repeated MaterializationInterval materialization_intervals = 3;
98102
}

‎sdk/python/feast/diff/registry_diff.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020
from feast.protos.feast.core.RequestFeatureView_pb2 import (
2121
RequestFeatureView as RequestFeatureViewProto,
2222
)
23+
from feast.protos.feast.core.StreamFeatureView_pb2 import (
24+
StreamFeatureView as StreamFeatureViewProto,
25+
)
2326
from feast.protos.feast.core.ValidationProfile_pb2 import (
2427
ValidationReference as ValidationReferenceProto,
2528
)
@@ -106,6 +109,7 @@ def tag_objects_for_keep_delete_update_add(
106109
FeatureServiceProto,
107110
OnDemandFeatureViewProto,
108111
RequestFeatureViewProto,
112+
StreamFeatureViewProto,
109113
ValidationReferenceProto,
110114
)
111115

@@ -292,6 +296,7 @@ def apply_diff_to_registry(
292296
FeastObjectType.FEATURE_VIEW,
293297
FeastObjectType.ON_DEMAND_FEATURE_VIEW,
294298
FeastObjectType.REQUEST_FEATURE_VIEW,
299+
FeastObjectType.STREAM_FEATURE_VIEW,
295300
]:
296301
feature_view_obj = cast(
297302
BaseFeatureView, feast_object_diff.current_feast_object
@@ -331,6 +336,7 @@ def apply_diff_to_registry(
331336
FeastObjectType.FEATURE_VIEW,
332337
FeastObjectType.ON_DEMAND_FEATURE_VIEW,
333338
FeastObjectType.REQUEST_FEATURE_VIEW,
339+
FeastObjectType.STREAM_FEATURE_VIEW,
334340
]:
335341
registry.apply_feature_view(
336342
cast(BaseFeatureView, feast_object_diff.new_feast_object),

‎sdk/python/feast/feature_store.py‎

Lines changed: 51 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -372,6 +372,40 @@ def _get_feature_view(
372372
feature_view.entities = []
373373
return feature_view
374374

375+
@log_exceptions_and_usage
376+
def get_stream_feature_view(
377+
self, name: str, allow_registry_cache: bool = False
378+
) -> StreamFeatureView:
379+
"""
380+
Retrieves a stream feature view.
381+
382+
Args:
383+
name: Name of stream feature view.
384+
allow_registry_cache: (Optional) Whether to allow returning this entity from a cached registry
385+
386+
Returns:
387+
The specified stream feature view.
388+
389+
Raises:
390+
FeatureViewNotFoundException: The feature view could not be found.
391+
"""
392+
return self._get_stream_feature_view(
393+
name, allow_registry_cache=allow_registry_cache
394+
)
395+
396+
def _get_stream_feature_view(
397+
self,
398+
name: str,
399+
hide_dummy_entity: bool = True,
400+
allow_registry_cache: bool = False,
401+
) -> StreamFeatureView:
402+
stream_feature_view = self._registry.get_stream_feature_view(
403+
name, self.project, allow_cache=allow_registry_cache
404+
)
405+
if hide_dummy_entity and stream_feature_view.entities[0] == DUMMY_ENTITY_NAME:
406+
stream_feature_view.entities = []
407+
return stream_feature_view
408+
375409
@log_exceptions_and_usage
376410
def get_on_demand_feature_view(self, name: str) -> OnDemandFeatureView:
377411
"""
@@ -935,7 +969,6 @@ def get_historical_features(
935969
all_feature_views,
936970
all_request_feature_views,
937971
all_on_demand_feature_views,
938-
all_stream_feature_views,
939972
) = self._get_feature_views_to_use(features)
940973

941974
if all_request_feature_views:
@@ -1321,9 +1354,14 @@ def write_to_online_store(
13211354
ingests data directly into the Online store
13221355
"""
13231356
# TODO: restrict this to work with online StreamFeatureViews and validate the FeatureView type
1324-
feature_view = self.get_feature_view(
1325-
feature_view_name, allow_registry_cache=allow_registry_cache
1326-
)
1357+
try:
1358+
feature_view = self.get_stream_feature_view(
1359+
feature_view_name, allow_registry_cache=allow_registry_cache
1360+
)
1361+
except FeatureViewNotFoundException:
1362+
feature_view = self.get_feature_view(
1363+
feature_view_name, allow_registry_cache=allow_registry_cache
1364+
)
13271365
entities = []
13281366
for entity_name in feature_view.entities:
13291367
entities.append(
@@ -1456,7 +1494,6 @@ def _get_online_features(
14561494
requested_feature_views,
14571495
requested_request_feature_views,
14581496
requested_on_demand_feature_views,
1459-
request_stream_feature_views,
14601497
) = self._get_feature_views_to_use(
14611498
features=features, allow_cache=True, hide_dummy_entity=False
14621499
)
@@ -1994,15 +2031,17 @@ def _get_feature_views_to_use(
19942031
allow_cache=False,
19952032
hide_dummy_entity: bool = True,
19962033
) -> Tuple[
1997-
List[FeatureView],
1998-
List[RequestFeatureView],
1999-
List[OnDemandFeatureView],
2000-
List[StreamFeatureView],
2034+
List[FeatureView], List[RequestFeatureView], List[OnDemandFeatureView],
20012035
]:
20022036

20032037
fvs = {
20042038
fv.name: fv
2005-
for fv in self._list_feature_views(allow_cache, hide_dummy_entity)
2039+
for fv in [
2040+
*self._list_feature_views(allow_cache, hide_dummy_entity),
2041+
*self._registry.list_stream_feature_views(
2042+
project=self.project, allow_cache=allow_cache
2043+
),
2044+
]
20062045
}
20072046

20082047
request_fvs = {
@@ -2019,15 +2058,8 @@ def _get_feature_views_to_use(
20192058
)
20202059
}
20212060

2022-
sfvs = {
2023-
fv.name: fv
2024-
for fv in self._registry.list_stream_feature_views(
2025-
project=self.project, allow_cache=allow_cache
2026-
)
2027-
}
2028-
20292061
if isinstance(features, FeatureService):
2030-
fvs_to_use, request_fvs_to_use, od_fvs_to_use, sfvs_to_use = [], [], [], []
2062+
fvs_to_use, request_fvs_to_use, od_fvs_to_use = [], [], []
20312063
for fv_name, projection in [
20322064
(projection.name, projection)
20332065
for projection in features.feature_view_projections
@@ -2048,23 +2080,18 @@ def _get_feature_views_to_use(
20482080
fv = fvs[projection.name].with_projection(copy.copy(projection))
20492081
if fv not in fvs_to_use:
20502082
fvs_to_use.append(fv)
2051-
elif fv_name in sfvs:
2052-
sfvs_to_use.append(
2053-
sfvs[fv_name].with_projection(copy.copy(projection))
2054-
)
20552083
else:
20562084
raise ValueError(
20572085
f"The provided feature service {features.name} contains a reference to a feature view"
20582086
f"{fv_name} which doesn't exist. Please make sure that you have created the feature view"
20592087
f'{fv_name} and that you have registered it by running "apply".'
20602088
)
2061-
views_to_use = (fvs_to_use, request_fvs_to_use, od_fvs_to_use, sfvs_to_use)
2089+
views_to_use = (fvs_to_use, request_fvs_to_use, od_fvs_to_use)
20622090
else:
20632091
views_to_use = (
20642092
[*fvs.values()],
20652093
[*request_fvs.values()],
20662094
[*od_fvs.values()],
2067-
[*sfvs.values()],
20682095
)
20692096

20702097
return views_to_use

‎sdk/python/feast/infra/online_stores/sqlite.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,7 +211,10 @@ def plan(
211211
path=self._get_db_path(config),
212212
name=_table_id(project, FeatureView.from_proto(view)),
213213
)
214-
for view in desired_registry_proto.feature_views
214+
for view in [
215+
*desired_registry_proto.feature_views,
216+
*desired_registry_proto.stream_feature_views,
217+
]
215218
]
216219
return infra_objects
217220

‎sdk/python/feast/registry.py‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@ class FeastObjectType(Enum):
7777
FEATURE_VIEW = "feature view"
7878
ON_DEMAND_FEATURE_VIEW = "on demand feature view"
7979
REQUEST_FEATURE_VIEW = "request feature view"
80+
STREAM_FEATURE_VIEW = "stream feature view"
8081
FEATURE_SERVICE = "feature service"
8182

8283
@staticmethod
@@ -93,6 +94,9 @@ def get_objects_from_registry(
9394
FeastObjectType.REQUEST_FEATURE_VIEW: registry.list_request_feature_views(
9495
project=project
9596
),
97+
FeastObjectType.STREAM_FEATURE_VIEW: registry.list_stream_feature_views(
98+
project=project,
99+
),
96100
FeastObjectType.FEATURE_SERVICE: registry.list_feature_services(
97101
project=project
98102
),
@@ -108,6 +112,7 @@ def get_objects_from_repo_contents(
108112
FeastObjectType.FEATURE_VIEW: repo_contents.feature_views,
109113
FeastObjectType.ON_DEMAND_FEATURE_VIEW: repo_contents.on_demand_feature_views,
110114
FeastObjectType.REQUEST_FEATURE_VIEW: repo_contents.request_feature_views,
115+
FeastObjectType.STREAM_FEATURE_VIEW: repo_contents.stream_feature_views,
111116
FeastObjectType.FEATURE_SERVICE: repo_contents.feature_services,
112117
}
113118

@@ -717,6 +722,30 @@ def get_feature_view(
717722
return FeatureView.from_proto(feature_view_proto)
718723
raise FeatureViewNotFoundException(name, project)
719724

725+
def get_stream_feature_view(
726+
self, name: str, project: str, allow_cache: bool = False
727+
) -> StreamFeatureView:
728+
"""
729+
Retrieves a stream feature view.
730+
731+
Args:
732+
name: Name of stream feature view
733+
project: Feast project that this stream feature view belongs to
734+
allow_cache: Allow returning feature view from the cached registry
735+
736+
Returns:
737+
Returns either the specified feature view, or raises an exception if
738+
none is found
739+
"""
740+
registry_proto = self._get_registry_proto(allow_cache=allow_cache)
741+
for feature_view_proto in registry_proto.stream_feature_views:
742+
if (
743+
feature_view_proto.spec.name == name
744+
and feature_view_proto.spec.project == project
745+
):
746+
return StreamFeatureView.from_proto(feature_view_proto)
747+
raise FeatureViewNotFoundException(name, project)
748+
720749
def delete_feature_service(self, name: str, project: str, commit: bool = True):
721750
"""
722751
Deletes a feature service or raises an exception if not found.

‎sdk/python/feast/repo_operations.py‎

Lines changed: 33 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,8 @@
1313
from click.exceptions import BadParameter
1414

1515
from feast import PushSource
16-
from feast.data_source import DataSource
16+
from feast.batch_feature_view import BatchFeatureView
17+
from feast.data_source import DataSource, KafkaSource
1718
from feast.diff.registry_diff import extract_objects_for_keep_delete_update_add
1819
from feast.entity import Entity
1920
from feast.feature_service import FeatureService
@@ -25,6 +26,7 @@
2526
from feast.repo_config import RepoConfig
2627
from feast.repo_contents import RepoContents
2728
from feast.request_feature_view import RequestFeatureView
29+
from feast.stream_feature_view import StreamFeatureView
2830
from feast.usage import log_exceptions_and_usage
2931

3032

@@ -122,8 +124,11 @@ def parse_repo(repo_root: Path) -> RepoContents:
122124
):
123125
res.data_sources.append(obj)
124126
data_sources_set.add(obj)
125-
if isinstance(obj, FeatureView) and not any(
126-
(obj is fv) for fv in res.feature_views
127+
if (
128+
isinstance(obj, FeatureView)
129+
and not any((obj is fv) for fv in res.feature_views)
130+
and not isinstance(obj, StreamFeatureView)
131+
and not isinstance(obj, BatchFeatureView)
127132
):
128133
res.feature_views.append(obj)
129134
if isinstance(obj.stream_source, PushSource) and not any(
@@ -133,6 +138,19 @@ def parse_repo(repo_root: Path) -> RepoContents:
133138
# Don't add if the push source's batch source is a duplicate of an existing batch source
134139
if push_source_dep not in data_sources_set:
135140
res.data_sources.append(push_source_dep)
141+
elif isinstance(obj, StreamFeatureView) and not any(
142+
(obj is sfv) for sfv in res.stream_feature_views
143+
):
144+
res.stream_feature_views.append(obj)
145+
if (
146+
isinstance(obj.stream_source, PushSource)
147+
or isinstance(obj.stream_source, KafkaSource)
148+
and not any((obj is ds) for ds in res.data_sources)
149+
):
150+
batch_source_dep = obj.stream_source.batch_source
151+
# Don't add if the push source's batch source is a duplicate of an existing batch source
152+
if batch_source_dep and batch_source_dep not in data_sources_set:
153+
res.data_sources.append(batch_source_dep)
136154
elif isinstance(obj, Entity) and not any(
137155
(obj is entity) for entity in res.entities
138156
):
@@ -196,7 +214,12 @@ def extract_objects_for_apply_delete(project, registry, repo):
196214

197215
all_to_apply: List[
198216
Union[
199-
Entity, FeatureView, RequestFeatureView, OnDemandFeatureView, FeatureService
217+
Entity,
218+
FeatureView,
219+
RequestFeatureView,
220+
OnDemandFeatureView,
221+
StreamFeatureView,
222+
FeatureService,
200223
]
201224
] = []
202225
for object_type in FEAST_OBJECT_TYPES:
@@ -205,7 +228,12 @@ def extract_objects_for_apply_delete(project, registry, repo):
205228

206229
all_to_delete: List[
207230
Union[
208-
Entity, FeatureView, RequestFeatureView, OnDemandFeatureView, FeatureService
231+
Entity,
232+
FeatureView,
233+
RequestFeatureView,
234+
OnDemandFeatureView,
235+
StreamFeatureView,
236+
FeatureService,
209237
]
210238
] = []
211239
for object_type in FEAST_OBJECT_TYPES:

0 commit comments

Comments
 (0)