fix: Forcing ODFV udfs to be __main__ module and fixing false positive duplicate data source warning (#2677)

* fix: Forcing ODFV udfs to be __main__ module so clients don't need the udf's module. Fixing duplicate data source conflict between push source and batch source

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

Author: adchia

sdk/python/feast/on_demand_feature_view.py

@@ -628,7 +628,14 @@ def on_demand_feature_view(
     if not _sources:
         raise ValueError("The `sources` parameter must be specified.")
 
+    def mainify(obj):
+        # Needed to allow dill to properly serialize the udf. Otherwise, clients will need to have a file with the same
+        # name as the original file defining the ODFV.
+        if obj.__module__ != "__main__":
+            obj.__module__ = "__main__"
+
     def decorator(user_function):
+        mainify(user_function)
         on_demand_feature_view_obj = OnDemandFeatureView(
             name=user_function.__name__,
             sources=_sources,

sdk/python/feast/repo_operations.py

@@ -110,6 +110,7 @@ def parse_repo(repo_root: Path) -> RepoContents:
         request_feature_views=[],
     )
 
+    data_sources_set = set()
     for repo_file in get_repo_files(repo_root):
         module_path = py_path_to_module(repo_file)
         module = importlib.import_module(module_path)
@@ -119,14 +120,18 @@ def parse_repo(repo_root: Path) -> RepoContents:
                 (obj is ds) for ds in res.data_sources
             ):
                 res.data_sources.append(obj)
+                data_sources_set.add(obj)
             if isinstance(obj, FeatureView) and not any(
                 (obj is fv) for fv in res.feature_views
             ):
                 res.feature_views.append(obj)
                 if isinstance(obj.stream_source, PushSource) and not any(
                     (obj is ds) for ds in res.data_sources
                 ):
-                    res.data_sources.append(obj.stream_source.batch_source)
+                    push_source_dep = obj.stream_source.batch_source
+                    # Don't add if the push source's batch source is a duplicate of an existing batch source
+                    if push_source_dep not in data_sources_set:
+                        res.data_sources.append(push_source_dep)
             elif isinstance(obj, Entity) and not any(
                 (obj is entity) for entity in res.entities
             ):