fix: Add inlined data sources to the top level registry

achals · achals · commit 40a5f6dbc073 · 2022-03-25T14:35:46.000-07:00
Signed-off-by: Achal Shah &lt;achals@gmail.com&gt;
diff --git 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
@@ -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(

Original file line number	Diff line number	Diff line change
`@@ -733,7 +733,7 @@ func groupFeatureRefs(requestedFeatureViews []*featureViewAndRefs,`
`733`	`733`	`joinKeys := make([]string, 0)`
`734`	`734`	`fv := featuresAndView.view`
`735`	`735`	`featureNames := featuresAndView.featureRefs`
`736`		`- for entity, _ := range fv.entities {`
	`736`	`+ for entity := range fv.entities {`
`737`	`737`	`joinKeys = append(joinKeys, entityNameToJoinKeyMap[entity])`
`738`	`738`	`}`
`739`	`739`