Merge changes from master

feast-dev · feast-ci-bot · Jul 22, 2021 · Jul 6, 2021 · Jul 6, 2021 · Jul 6, 2021
commit 6f096867b6d8a685443cc213c11d5fb8a2131c5c
@@ -23,8 +23,8 @@
 
 from feast import utils
 from feast.entity import Entity
-from feast.errors import FeastProviderLoginError, FeatureViewNotFoundException
 from feast.feature_service import FeatureService
+from feast.errors import FeatureNameCollisionError, FeatureViewNotFoundException
 from feast.feature_view import FeatureView
 from feast.inference import (
     update_data_sources_with_inferred_event_timestamp_col,
@@ -304,6 +304,7 @@ def get_historical_features(
         entity_df: Union[pd.DataFrame, str],
         features: Union[List[str], FeatureService],
         feature_refs: Optional[List[str]] = None,
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         """Enrich an entity dataframe with historical feature values for either training or batch scoring.
 
@@ -346,10 +347,11 @@ def get_historical_features(
             >>> model.fit(feature_data) # insert your modeling framework here.
         """
         all_feature_views = self._registry.list_feature_views(project=self.project)
-        try:
-            feature_views = _get_requested_feature_views(features, all_feature_views)
-        except FeatureViewNotFoundException as e:
-            sys.exit(e)
+
+        _validate_feature_refs(feature_refs, full_feature_names)
+        feature_views = list(
+            view for view, _ in _group_feature_refs(feature_refs, all_feature_views)
+        )
 
         _features = features or feature_refs
         if not _features:
@@ -365,17 +367,16 @@ def get_historical_features(
             _feature_refs = _features
 
         provider = self._get_provider()
-        try:
-            job = provider.get_historical_features(
-                self.config,
-                feature_views,
-                _feature_refs,
-                entity_df,
-                self._registry,
-                self.project,
-            )
-        except FeastProviderLoginError as e:
-            sys.exit(e)
+
+        job = provider.get_historical_features(
+            self.config,
+            feature_views,
+            feature_refs,
+            entity_df,
+            self._registry,
+            self.project,
+            full_feature_names,
+        )
 
         return job
 
@@ -545,6 +546,7 @@ def get_online_features(
         features: Union[List[str], FeatureService],
         entity_rows: List[Dict[str, Any]],
         feature_refs: Optional[List[str]] = None,
+        full_feature_names: bool = False,
     ) -> OnlineResponse:
         """
         Retrieves the latest online feature data.
@@ -620,7 +622,8 @@ def get_online_features(
             project=self.project, allow_cache=True
         )
 
-        grouped_refs = _group_refs(_features, all_feature_views)
+        _validate_feature_refs(feature_refs, full_feature_names)
+        grouped_refs = _group_feature_refs(feature_refs, all_feature_views)
         for table, requested_features in grouped_refs:
             entity_keys = _get_table_entity_keys(
                 table, union_of_entity_keys, entity_name_to_join_key_map
@@ -679,8 +682,32 @@ def _entity_row_to_field_values(
     return result
 
 
-def _group_refs(
-    features: Union[List[str], FeatureService], all_feature_views: List[FeatureView],
+def _validate_feature_refs(feature_refs: Union[List[str], FeatureService], full_feature_names: bool = False):
+    collided_feature_refs = []
+
+    if full_feature_names:
+        collided_feature_refs = [
+            ref for ref, occurrences in Counter(feature_refs).items() if occurrences > 1
+        ]
+    else:
+        feature_names = [ref.split(":")[1] for ref in feature_refs]
+        collided_feature_names = [
+            ref
+            for ref, occurrences in Counter(feature_names).items()
+            if occurrences > 1
+        ]
+
+        for feature_name in collided_feature_names:
+            collided_feature_refs.extend(
+                [ref for ref in feature_refs if ref.endswith(":" + feature_name)]
+            )
+
+    if len(collided_feature_refs) > 0:
+        raise FeatureNameCollisionError(collided_feature_refs, full_feature_names)
+
+
+def _group_feature_refs(
+    features: Union[List[str], FeatureService], all_feature_views: List[FeatureView]
 ) -> List[Tuple[FeatureView, List[str]]]:
     """ Get list of feature views and corresponding feature names based on feature references"""
 
@@ -714,14 +741,6 @@ def _get_features_refs_from_feature_services(
     return [f"{feature_service.name}:{f.name}" for f in feature_service.features]
 
 
-def _get_requested_feature_views(
-    features: Union[List[str], FeatureService], all_feature_views: List[FeatureView],
-) -> List[FeatureView]:
-    """Get list of feature views based on feature references"""
-    # TODO: Get rid of this function. We only need _group_refs
-    return list(view for view, _ in _group_refs(features, all_feature_views))
-
-
 def _get_table_entity_keys(
     table: FeatureView, entity_keys: List[EntityKeyProto], join_key_map: Dict[str, str],
 ) -> List[EntityKeyProto]:

diff --git a/sdk/python/tests/test_offline_online_store_consistency.py b/sdk/python/tests/test_offline_online_store_consistency.py
@@ -254,6 +254,7 @@ def check_offline_and_online_features(
             {"driver_id": [driver_id], "event_timestamp": [event_timestamp]}
         ),
         features=[f"{fv.name}:value"],
+        full_feature_names=full_feature_names,
     ).to_df()
 
     if full_feature_names:

diff --git a/sdk/python/tests/test_online_retrieval.py b/sdk/python/tests/test_online_retrieval.py
@@ -126,7 +126,9 @@ def test_online() -> None:
         # invalid table reference
         with pytest.raises(FeatureViewNotFoundException):
             store.get_online_features(
-                features=["driver_locations_bad:lon"], entity_rows=[{"driver": 1}],
+                features=["driver_locations_bad:lon"],
+                entity_rows=[{"driver": 1}],
+                full_feature_names=False,
             )
 
         # Create new FeatureStore object with fast cache invalidation