Skip to content

Commit ef9e0bb

Browse files
franciscojavierarceofranciscojavierarceo
authored
feat: Adding write capability to online store to on demand feature views (#4585)
* merged changes Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * saving progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merged changes to odfv Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * linted Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding the test needed to show the expected behavior Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated test case Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * saving progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merging Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merged Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merged Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merging Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding the entity keys for now to do retrieval Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding entity to odfv Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * checking in progress...getting closer Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * may have to revert some of this...looks like the challenge is getting the entities correct when storing writes. just checking in progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * moving things around to make it easier to debug Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * debugging Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merged Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merging Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * Rebasing and merging changes from other PR Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * Merging changes continued Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * update the _make_inference to include odfvs with writes in the update map Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * have the table being written now...the create table happens in the SqliteOnlineStore.update() method Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * checking in progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding logs Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updating permissions Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * going to error out on purpose Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding unit test and merging changes Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * almost got everything working and type validation behaving Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * cleaned up and have tests behaving Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding print Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * removing print Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * checking in progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updating test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * linted and updated Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * removed print Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated tests to test actual behavior Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * checking in progress Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * changing typo Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updating test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * testing changes Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * checking to see if thing still working Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * removed print Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * undo change for odfv file Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated tests Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * okay well have the unit test working Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * type changes, hope i dont regret them Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated stream feature view piece Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated sfv ifelse Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * removing print Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * formatted and updated test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * resolving some linter errors Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * fixed linter and formatting Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * okay think it is working Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * linter Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated type map for integration tests Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated local feature store test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * fixed local fs test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * chore: Updated snowflake test to be more explicit about post apply entity_columns return value (#4603) chore: updated snowflake test to be more explicit about post apply entity_column return value Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * merging Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * fixed test to entity_rows_to_read Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * resolved inf conflicts Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * lint Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * Updated tests and lint, think I have everything working Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> --------- Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
1 parent b5ab6c7 commit ef9e0bb

20 files changed

+737
-107
lines changed

‎sdk/python/feast/errors.py‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -467,10 +467,15 @@ def __init__(self):
467467

468468

469469
class DataFrameSerializationError(FeastError):
470-
def __init__(self, input_dict: dict):
471-
super().__init__(
472-
f"Failed to serialize the provided dictionary into a pandas DataFrame: {input_dict.keys()}"
473-
)
470+
def __init__(self, input: Any):
471+
if isinstance(input, dict):
472+
super().__init__(
473+
f"Failed to serialize the provided dictionary into a pandas DataFrame: {input.keys()}"
474+
)
475+
else:
476+
super().__init__(
477+
"Failed to serialize the provided input into a pandas DataFrame"
478+
)
474479

475480

476481
class PermissionNotFoundException(FeastError):

‎sdk/python/feast/feature_service.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,9 @@ def __init__(
8484
if isinstance(feature_grouping, BaseFeatureView):
8585
self.feature_view_projections.append(feature_grouping.projection)
8686

87-
def infer_features(self, fvs_to_update: Dict[str, FeatureView]):
87+
def infer_features(
88+
self, fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]]
89+
):
8890
"""
8991
Infers the features for the projections of this feature service, and updates this feature
9092
service in place.

‎sdk/python/feast/feature_store.py‎

Lines changed: 54 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -608,7 +608,12 @@ def _make_inferences(
608608
update_feature_views_with_inferred_features_and_entities(
609609
sfvs_to_update, entities + entities_to_update, self.config
610610
)
611-
# TODO(kevjumba): Update schema inferrence
611+
# We need to attach the time stamp fields to the underlying data sources
612+
# and cascade the dependencies
613+
update_feature_views_with_inferred_features_and_entities(
614+
odfvs_to_update, entities + entities_to_update, self.config
615+
)
616+
# TODO(kevjumba): Update schema inference
612617
for sfv in sfvs_to_update:
613618
if not sfv.schema:
614619
raise ValueError(
@@ -618,8 +623,13 @@ def _make_inferences(
618623
for odfv in odfvs_to_update:
619624
odfv.infer_features()
620625

626+
odfvs_to_write = [
627+
odfv for odfv in odfvs_to_update if odfv.write_to_online_store
628+
]
629+
# Update to include ODFVs with write to online store
621630
fvs_to_update_map = {
622-
view.name: view for view in [*views_to_update, *sfvs_to_update]
631+
view.name: view
632+
for view in [*views_to_update, *sfvs_to_update, *odfvs_to_write]
623633
}
624634
for feature_service in feature_services_to_update:
625635
feature_service.infer_features(fvs_to_update=fvs_to_update_map)
@@ -847,6 +857,11 @@ def apply(
847857
]
848858
sfvs_to_update = [ob for ob in objects if isinstance(ob, StreamFeatureView)]
849859
odfvs_to_update = [ob for ob in objects if isinstance(ob, OnDemandFeatureView)]
860+
odfvs_with_writes_to_update = [
861+
ob
862+
for ob in objects
863+
if isinstance(ob, OnDemandFeatureView) and ob.write_to_online_store
864+
]
850865
services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)]
851866
data_sources_set_to_update = {
852867
ob for ob in objects if isinstance(ob, DataSource)
@@ -868,10 +883,23 @@ def apply(
868883
for batch_source in batch_sources_to_add:
869884
data_sources_set_to_update.add(batch_source)
870885

871-
for fv in itertools.chain(views_to_update, sfvs_to_update):
872-
data_sources_set_to_update.add(fv.batch_source)
873-
if fv.stream_source:
874-
data_sources_set_to_update.add(fv.stream_source)
886+
for fv in itertools.chain(
887+
views_to_update, sfvs_to_update, odfvs_with_writes_to_update
888+
):
889+
if isinstance(fv, FeatureView):
890+
data_sources_set_to_update.add(fv.batch_source)
891+
if hasattr(fv, "stream_source"):
892+
if fv.stream_source:
893+
data_sources_set_to_update.add(fv.stream_source)
894+
if isinstance(fv, OnDemandFeatureView):
895+
for source_fvp in fv.source_feature_view_projections:
896+
odfv_batch_source: Optional[DataSource] = (
897+
fv.source_feature_view_projections[source_fvp].batch_source
898+
)
899+
if odfv_batch_source is not None:
900+
data_sources_set_to_update.add(odfv_batch_source)
901+
else:
902+
pass
875903

876904
for odfv in odfvs_to_update:
877905
for v in odfv.source_request_sources.values():
@@ -884,7 +912,9 @@ def apply(
884912

885913
# Validate all feature views and make inferences.
886914
self._validate_all_feature_views(
887-
views_to_update, odfvs_to_update, sfvs_to_update
915+
views_to_update,
916+
odfvs_to_update,
917+
sfvs_to_update,
888918
)
889919
self._make_inferences(
890920
data_sources_to_update,
@@ -989,7 +1019,9 @@ def apply(
9891019
tables_to_delete: List[FeatureView] = (
9901020
views_to_delete + sfvs_to_delete if not partial else [] # type: ignore
9911021
)
992-
tables_to_keep: List[FeatureView] = views_to_update + sfvs_to_update # type: ignore
1022+
tables_to_keep: List[
1023+
Union[FeatureView, StreamFeatureView, OnDemandFeatureView]
1024+
] = views_to_update + sfvs_to_update + odfvs_with_writes_to_update # type: ignore
9931025

9941026
self._get_provider().update_infra(
9951027
project=self.project,
@@ -1444,19 +1476,18 @@ def write_to_online_store(
14441476
inputs: Optional the dictionary object to be written
14451477
allow_registry_cache (optional): Whether to allow retrieving feature views from a cached registry.
14461478
"""
1447-
# TODO: restrict this to work with online StreamFeatureViews and validate the FeatureView type
1479+
feature_view_dict = {
1480+
fv_proto.name: fv_proto
1481+
for fv_proto in self.list_all_feature_views(allow_registry_cache)
1482+
}
14481483
try:
1449-
feature_view: FeatureView = self.get_stream_feature_view(
1450-
feature_view_name, allow_registry_cache=allow_registry_cache
1451-
)
1484+
feature_view = feature_view_dict[feature_view_name]
14521485
except FeatureViewNotFoundException:
1453-
feature_view = self.get_feature_view(
1454-
feature_view_name, allow_registry_cache=allow_registry_cache
1455-
)
1486+
raise FeatureViewNotFoundException(feature_view_name, self.project)
14561487
if df is not None and inputs is not None:
14571488
raise ValueError("Both df and inputs cannot be provided at the same time.")
14581489
if df is None and inputs is not None:
1459-
if isinstance(inputs, dict):
1490+
if isinstance(inputs, dict) or isinstance(inputs, List):
14601491
try:
14611492
df = pd.DataFrame(inputs)
14621493
except Exception as _:
@@ -1465,6 +1496,13 @@ def write_to_online_store(
14651496
pass
14661497
else:
14671498
raise ValueError("inputs must be a dictionary or a pandas DataFrame.")
1499+
if df is not None and inputs is None:
1500+
if isinstance(df, dict) or isinstance(df, List):
1501+
try:
1502+
df = pd.DataFrame(df)
1503+
except Exception as _:
1504+
raise DataFrameSerializationError(df)
1505+
14681506
provider = self._get_provider()
14691507
provider.ingest_df(feature_view, df)
14701508

‎sdk/python/feast/inference.py‎

Lines changed: 23 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -183,10 +183,13 @@ def update_feature_views_with_inferred_features_and_entities(
183183
)
184184

185185
if not fv.features:
186-
raise RegistryInferenceFailure(
187-
"FeatureView",
188-
f"Could not infer Features for the FeatureView named {fv.name}.",
189-
)
186+
if isinstance(fv, OnDemandFeatureView):
187+
return None
188+
else:
189+
raise RegistryInferenceFailure(
190+
"FeatureView",
191+
f"Could not infer Features for the FeatureView named {fv.name}.",
192+
)
190193

191194

192195
def _infer_features_and_entities(
@@ -209,6 +212,7 @@ def _infer_features_and_entities(
209212
fv, join_keys, run_inference_for_features, config
210213
)
211214

215+
entity_columns: List[Field] = fv.entity_columns if fv.entity_columns else []
212216
columns_to_exclude = {
213217
fv.batch_source.timestamp_field,
214218
fv.batch_source.created_timestamp_column,
@@ -235,7 +239,7 @@ def _infer_features_and_entities(
235239
if field.name not in [
236240
entity_column.name for entity_column in fv.entity_columns
237241
]:
238-
fv.entity_columns.append(field)
242+
entity_columns.append(field)
239243
elif not re.match(
240244
"^__|__$", col_name
241245
): # double underscores often signal an internal-use column
@@ -256,6 +260,8 @@ def _infer_features_and_entities(
256260
if field.name not in [feature.name for feature in fv.features]:
257261
fv.features.append(field)
258262

263+
fv.entity_columns = entity_columns
264+
259265

260266
def _infer_on_demand_features_and_entities(
261267
fv: OnDemandFeatureView,
@@ -282,18 +288,19 @@ def _infer_on_demand_features_and_entities(
282288

283289
batch_source = getattr(source_feature_view, "batch_source")
284290
batch_field_mapping = getattr(batch_source or None, "field_mapping")
285-
if batch_field_mapping:
286-
for (
287-
original_col,
288-
mapped_col,
289-
) in batch_field_mapping.items():
290-
if mapped_col in columns_to_exclude:
291-
columns_to_exclude.remove(mapped_col)
292-
columns_to_exclude.add(original_col)
291+
for (
292+
original_col,
293+
mapped_col,
294+
) in batch_field_mapping.items():
295+
if mapped_col in columns_to_exclude:
296+
columns_to_exclude.remove(mapped_col)
297+
columns_to_exclude.add(original_col)
298+
299+
table_column_names_and_types = batch_source.get_table_column_names_and_types(
300+
config
301+
)
302+
batch_field_mapping = getattr(batch_source, "field_mapping", {})
293303

294-
table_column_names_and_types = (
295-
batch_source.get_table_column_names_and_types(config)
296-
)
297304
for col_name, col_datatype in table_column_names_and_types:
298305
if col_name in columns_to_exclude:
299306
continue

‎sdk/python/feast/infra/materialization/aws_lambda/lambda_engine.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
from feast.infra.offline_stores.offline_store import OfflineStore
2525
from feast.infra.online_stores.online_store import OnlineStore
2626
from feast.infra.registry.base_registry import BaseRegistry
27+
from feast.on_demand_feature_view import OnDemandFeatureView
2728
from feast.repo_config import FeastConfigBaseModel, RepoConfig
2829
from feast.stream_feature_view import StreamFeatureView
2930
from feast.utils import _get_column_names
@@ -80,10 +81,10 @@ def update(
8081
self,
8182
project: str,
8283
views_to_delete: Sequence[
83-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
84+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
8485
],
8586
views_to_keep: Sequence[
86-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
87+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
8788
],
8889
entities_to_delete: Sequence[Entity],
8990
entities_to_keep: Sequence[Entity],

‎sdk/python/feast/infra/materialization/batch_materialization_engine.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
from feast.infra.offline_stores.offline_store import OfflineStore
1313
from feast.infra.online_stores.online_store import OnlineStore
1414
from feast.infra.registry.base_registry import BaseRegistry
15+
from feast.on_demand_feature_view import OnDemandFeatureView
1516
from feast.repo_config import RepoConfig
1617
from feast.stream_feature_view import StreamFeatureView
1718

@@ -89,7 +90,7 @@ def update(
8990
Union[BatchFeatureView, StreamFeatureView, FeatureView]
9091
],
9192
views_to_keep: Sequence[
92-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
93+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
9394
],
9495
entities_to_delete: Sequence[Entity],
9596
entities_to_keep: Sequence[Entity],

‎sdk/python/feast/infra/materialization/contrib/spark/spark_materialization_engine.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
from feast.infra.online_stores.online_store import OnlineStore
2424
from feast.infra.passthrough_provider import PassthroughProvider
2525
from feast.infra.registry.base_registry import BaseRegistry
26+
from feast.on_demand_feature_view import OnDemandFeatureView
2627
from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
2728
from feast.repo_config import FeastConfigBaseModel, RepoConfig
2829
from feast.stream_feature_view import StreamFeatureView
@@ -77,10 +78,10 @@ def update(
7778
self,
7879
project: str,
7980
views_to_delete: Sequence[
80-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
81+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
8182
],
8283
views_to_keep: Sequence[
83-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
84+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
8485
],
8586
entities_to_delete: Sequence[Entity],
8687
entities_to_keep: Sequence[Entity],

‎sdk/python/feast/infra/materialization/kubernetes/k8s_materialization_engine.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
from feast.infra.offline_stores.offline_store import OfflineStore
2525
from feast.infra.online_stores.online_store import OnlineStore
2626
from feast.infra.registry.base_registry import BaseRegistry
27+
from feast.on_demand_feature_view import OnDemandFeatureView
2728
from feast.repo_config import FeastConfigBaseModel
2829
from feast.stream_feature_view import StreamFeatureView
2930
from feast.utils import _get_column_names
@@ -122,10 +123,10 @@ def update(
122123
self,
123124
project: str,
124125
views_to_delete: Sequence[
125-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
126+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
126127
],
127128
views_to_keep: Sequence[
128-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
129+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
129130
],
130131
entities_to_delete: Sequence[Entity],
131132
entities_to_keep: Sequence[Entity],

‎sdk/python/feast/infra/materialization/local_engine.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
from feast.infra.offline_stores.offline_store import OfflineStore
1111
from feast.infra.online_stores.online_store import OnlineStore
1212
from feast.infra.registry.base_registry import BaseRegistry
13+
from feast.on_demand_feature_view import OnDemandFeatureView
1314
from feast.repo_config import FeastConfigBaseModel, RepoConfig
1415
from feast.stream_feature_view import StreamFeatureView
1516
from feast.utils import (
@@ -69,10 +70,10 @@ def update(
6970
self,
7071
project: str,
7172
views_to_delete: Sequence[
72-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
73+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
7374
],
7475
views_to_keep: Sequence[
75-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
76+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
7677
],
7778
entities_to_delete: Sequence[Entity],
7879
entities_to_keep: Sequence[Entity],

‎sdk/python/feast/infra/materialization/snowflake_engine.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
get_snowflake_online_store_path,
3232
package_snowpark_zip,
3333
)
34+
from feast.on_demand_feature_view import OnDemandFeatureView
3435
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
3536
from feast.protos.feast.types.Value_pb2 import Value as ValueProto
3637
from feast.repo_config import FeastConfigBaseModel, RepoConfig
@@ -120,10 +121,10 @@ def update(
120121
self,
121122
project: str,
122123
views_to_delete: Sequence[
123-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
124+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
124125
],
125126
views_to_keep: Sequence[
126-
Union[BatchFeatureView, StreamFeatureView, FeatureView]
127+
Union[BatchFeatureView, StreamFeatureView, FeatureView, OnDemandFeatureView]
127128
],
128129
entities_to_delete: Sequence[Entity],
129130
entities_to_keep: Sequence[Entity],

0 commit comments

Comments
 (0)