From 40a5f6dbc073d06d110cf5a2a850aece4f3f526a Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 25 Mar 2022 14:30:00 -0700 Subject: [PATCH] fix: Add inlined data sources to the top level registry Signed-off-by: Achal Shah --- go/internal/feast/featurestore.go | 2 +- sdk/python/feast/feature_store.py | 29 ++++++++++++++++++----------- 2 files changed, 19 insertions(+), 12 deletions(-) diff --git a/go/internal/feast/featurestore.go b/go/internal/feast/featurestore.go index c5b3a14eb81..492c158e52e 100644 --- a/go/internal/feast/featurestore.go +++ b/go/internal/feast/featurestore.go @@ -733,7 +733,7 @@ func groupFeatureRefs(requestedFeatureViews []*featureViewAndRefs, joinKeys := make([]string, 0) fv := featuresAndView.view featureNames := featuresAndView.featureRefs - for entity, _ := range fv.entities { + for entity := range fv.entities { joinKeys = append(joinKeys, entityNameToJoinKeyMap[entity]) } diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index b7214d9da8c..d6119d05f8b 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -560,7 +560,7 @@ def _plan( new_infra_proto = new_infra.to_proto() infra_diff = diff_infra_protos(current_infra_proto, new_infra_proto) - return (registry_diff, infra_diff, new_infra) + return registry_diff, infra_diff, new_infra @log_exceptions_and_usage def _apply_diffs( @@ -648,16 +648,23 @@ def apply( ] odfvs_to_update = [ob for ob in objects if isinstance(ob, OnDemandFeatureView)] services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)] - data_sources_to_update = [ob for ob in objects if isinstance(ob, DataSource)] - - if len(entities_to_update) + len(views_to_update) + len( - request_views_to_update - ) + len(odfvs_to_update) + len(services_to_update) + len( - data_sources_to_update - ) != len( - objects - ): - raise ValueError("Unknown object type provided as part of apply() call") + data_sources_set_to_update = { + ob for ob in objects if isinstance(ob, DataSource) + } + + for fv in views_to_update: + data_sources_set_to_update.add(fv.batch_source) + if fv.stream_source: + data_sources_set_to_update.add(fv.stream_source) + + for rfv in request_views_to_update: + data_sources_set_to_update.add(rfv.request_data_source) + + for odfv in odfvs_to_update: + for v in odfv.input_request_data_sources.values(): + data_sources_set_to_update.add(v) + + data_sources_to_update = list(data_sources_set_to_update) # Validate all feature views and make inferences. self._validate_all_feature_views(