feast-dev · feast-ci-bot · Jul 8, 2021 · Jun 12, 2021 · Jun 16, 2021 · Jun 16, 2021
diff --git a/README.md b/README.md
@@ -75,11 +75,11 @@ print(training_df.head())
 # model = ml.fit(training_df)
 ```
 ```commandline
-      event_timestamp  driver_id  driver_hourly_stats__conv_rate  driver_hourly_stats__acc_rate
-  2021-04-12 08:12:10       1002                        0.497279                       0.357702
-  2021-04-12 10:59:42       1001                        0.979747                       0.008166
-  2021-04-12 15:01:12       1004                        0.151432                       0.551748
-  2021-04-12 16:40:26       1003                        0.951506                       0.753572
+            event_timestamp  driver_id  conv_rate  acc_rate  avg_daily_trips
+0 2021-04-12 08:12:10+00:00       1002   0.713465  0.597095              531
+1 2021-04-12 10:59:42+00:00       1001   0.072752  0.044344               11
+2 2021-04-12 15:01:12+00:00       1004   0.658182  0.079150              220
+3 2021-04-12 16:40:26+00:00       1003   0.162092  0.309035              959
 
 ```
 

@@ -119,9 +119,9 @@ pprint(feature_vector)
 ```python
 {
     'driver_id': [1001],
-    'driver_hourly_stats__conv_rate': [0.49274],
-    'driver_hourly_stats__acc_rate': [0.92743],
-    'driver_hourly_stats__avg_daily_trips': [72],
+    'conv_rate': [0.49274],
+    'acc_rate': [0.92743],
+    'avg_daily_trips': [72],
 }
 ```
 

@@ -1,4 +1,4 @@
-from typing import Set
+from typing import List, Set
 
 from colorama import Fore, Style
 
@@ -88,6 +88,27 @@ def __init__(self, offline_store_name: str, data_source_name: str):
         )
 
 
+class FeatureNameCollisionError(Exception):
+    def __init__(self, feature_refs_collisions: List[str], full_feature_names):
+        if full_feature_names:
+            collisions = [ref.replace(":", "__") for ref in feature_refs_collisions]
+            error_message = (
+                "To resolve this collision, please ensure that the features in question "
+                "have different names."
+            )
+        else:
+            collisions = [ref.split(":")[1] for ref in feature_refs_collisions]
+            error_message = (
+                "To resolve this collision, either use the full feature name by setting "
+                "'full_feature_names=True', or ensure that the features in question have different names."
+            )
+
+        feature_names = ", ".join(set(collisions))
+        super().__init__(
+            f"Duplicate features named {feature_names} found.\n{error_message}"
+        )
+
+
 class FeastOnlineStoreInvalidName(Exception):
     def __init__(self, online_store_class_name: str):
         super().__init__(

@@ -12,8 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import os
-import sys
-from collections import OrderedDict, defaultdict
+from collections import Counter, OrderedDict, defaultdict
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
@@ -24,7 +23,7 @@
 
 from feast import utils
 from feast.entity import Entity
-from feast.errors import FeastProviderLoginError, FeatureViewNotFoundException
+from feast.errors import FeatureNameCollisionError, FeatureViewNotFoundException
 from feast.feature_view import FeatureView
 from feast.inference import (
     update_data_sources_with_inferred_event_timestamp_col,
@@ -230,9 +229,11 @@ def apply(
         update_entities_with_inferred_types_from_feature_views(
             entities_to_update, views_to_update, self.config
         )
+
         update_data_sources_with_inferred_event_timestamp_col(
             [view.input for view in views_to_update], self.config
         )
+
         for view in views_to_update:
             view.infer_features_from_input_source(self.config)
 
@@ -255,7 +256,10 @@ def apply(
 
     @log_exceptions_and_usage
     def get_historical_features(
-        self, entity_df: Union[pd.DataFrame, str], feature_refs: List[str],
+        self,
+        entity_df: Union[pd.DataFrame, str],
+        feature_refs: List[str],
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         """Enrich an entity dataframe with historical feature values for either training or batch scoring.
 
@@ -277,6 +281,9 @@ def get_historical_features(
                 SQL query. The query must be of a format supported by the configured offline store (e.g., BigQuery)
             feature_refs: A list of features that should be retrieved from the offline store. Feature references are of
                 the format "feature_view:feature", e.g., "customer_fv:daily_transactions".
+            full_feature_names: A boolean that provides the option to add the feature view prefixes to the feature names,
+                changing them from the format "feature" to "feature_view__feature" (e.g., "daily_transactions" changes to
+                "customer_fv__daily_transactions"). By default, this value is set to False.
 
         Returns:
             RetrievalJob which can be used to materialize the results.
@@ -289,32 +296,29 @@ def get_historical_features(
             >>> fs = FeatureStore(config=RepoConfig(provider="gcp"))
             >>> retrieval_job = fs.get_historical_features(
             >>>     entity_df="SELECT event_timestamp, order_id, customer_id from gcp_project.my_ds.customer_orders",
-            >>>     feature_refs=["customer:age", "customer:avg_orders_1d", "customer:avg_orders_7d"]
-            >>> )
+            >>>     feature_refs=["customer:age", "customer:avg_orders_1d", "customer:avg_orders_7d"],
+            >>>     )
             >>> feature_data = retrieval_job.to_df()
             >>> 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(
-                feature_refs, 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)
+        )
 
         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
 
@@ -480,7 +484,10 @@ def tqdm_builder(length):
 
     @log_exceptions_and_usage
     def get_online_features(
-        self, feature_refs: List[str], entity_rows: List[Dict[str, Any]],
+        self,
+        feature_refs: List[str],
+        entity_rows: List[Dict[str, Any]],
+        full_feature_names: bool = False,
     ) -> OnlineResponse:
         """
         Retrieves the latest online feature data.
@@ -548,7 +555,8 @@ def get_online_features(
             project=self.project, allow_cache=True
         )
 
-        grouped_refs = _group_refs(feature_refs, 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
@@ -565,13 +573,21 @@ def get_online_features(
 
                 if feature_data is None:
                     for feature_name in requested_features:
-                        feature_ref = f"{table.name}__{feature_name}"
+                        feature_ref = (
+                            f"{table.name}__{feature_name}"
+                            if full_feature_names
+                            else feature_name
+                        )
                         result_row.statuses[
                             feature_ref
                         ] = GetOnlineFeaturesResponse.FieldStatus.NOT_FOUND
                 else:
                     for feature_name in feature_data:
-                        feature_ref = f"{table.name}__{feature_name}"
+                        feature_ref = (
+                            f"{table.name}__{feature_name}"
+                            if full_feature_names
+                            else feature_name
+                        )
                         if feature_name in requested_features:
                             result_row.fields[feature_ref].CopyFrom(
                                 feature_data[feature_name]
@@ -599,7 +615,31 @@ def _entity_row_to_field_values(
     return result
 
 
-def _group_refs(
+def _validate_feature_refs(feature_refs: List[str], 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(
     feature_refs: List[str], all_feature_views: List[FeatureView]
 ) -> List[Tuple[FeatureView, List[str]]]:
     """ Get list of feature views and corresponding feature names based on feature references"""
@@ -612,6 +652,7 @@ def _group_refs(
 
     for ref in feature_refs:
         view_name, feat_name = ref.split(":")
+
         if view_name not in view_index:
             raise FeatureViewNotFoundException(view_name)
         views_features[view_name].append(feat_name)
@@ -622,14 +663,6 @@ def _group_refs(
     return result
 
 
-def _get_requested_feature_views(
-    feature_refs: List[str], 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(feature_refs, 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/feast/feature_view.py b/sdk/python/feast/feature_view.py
@@ -182,6 +182,8 @@ def to_proto(self) -> FeatureViewProto:
             ttl_duration = Duration()
             ttl_duration.FromTimedelta(self.ttl)
 
+        print(f"Stream soruce: {self.stream_source}, {type(self.stream_source)}")
+
         spec = FeatureViewSpecProto(
             name=self.name,
             entities=self.entities,

@@ -129,6 +129,7 @@ def get_historical_features(
         entity_df: Union[pandas.DataFrame, str],
         registry: Registry,
         project: str,
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         job = self.offline_store.get_historical_features(
             config=config,
@@ -137,5 +138,6 @@ def get_historical_features(
             entity_df=entity_df,
             registry=registry,
             project=project,
+            full_feature_names=full_feature_names,
         )
         return job
@@ -131,6 +131,7 @@ def get_historical_features(
         entity_df: Union[pandas.DataFrame, str],
         registry: Registry,
         project: str,
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         job = self.offline_store.get_historical_features(
             config=config,
@@ -139,5 +140,6 @@ def get_historical_features(
             entity_df=entity_df,
             registry=registry,
             project=project,
+            full_feature_names=full_feature_names,
         )
         return job
@@ -130,6 +130,7 @@ def get_historical_features(
         entity_df: Union[pd.DataFrame, str],
         registry: Registry,
         project: str,
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         return self.offline_store.get_historical_features(
             config=config,
@@ -138,6 +139,7 @@ def get_historical_features(
             entity_df=entity_df,
             registry=registry,
             project=project,
+            full_feature_names=full_feature_names,
         )
 
 

@@ -95,6 +95,7 @@ def get_historical_features(
         entity_df: Union[pandas.DataFrame, str],
         registry: Registry,
         project: str,
+        full_feature_names: bool = False,
     ) -> RetrievalJob:
         # TODO: Add entity_df validation in order to fail before interacting with BigQuery
         assert isinstance(config.offline_store, BigQueryOfflineStoreConfig)
@@ -121,7 +122,7 @@ def get_historical_features(
 
         # Build a query context containing all information required to template the BigQuery SQL query
         query_context = get_feature_view_query_context(
-            feature_refs, feature_views, registry, project
+            feature_refs, feature_views, registry, project, full_feature_names
         )
 
         # TODO: Infer min_timestamp and max_timestamp from entity_df
@@ -132,6 +133,7 @@ def get_historical_features(
             max_timestamp=datetime.now() + timedelta(days=1),
             left_table_query_string=str(table.reference),
             entity_df_event_timestamp_col=entity_df_event_timestamp_col,
+            full_feature_names=full_feature_names,
         )
 
         job = BigQueryRetrievalJob(query=query, client=client, config=config)
@@ -373,6 +375,7 @@ def get_feature_view_query_context(
     feature_views: List[FeatureView],
     registry: Registry,
     project: str,
+    full_feature_names: bool = False,
 ) -> List[FeatureViewQueryContext]:
     """Build a query context containing all information required to template a BigQuery point-in-time SQL query"""
 
@@ -432,6 +435,7 @@ def build_point_in_time_query(
     max_timestamp: datetime,
     left_table_query_string: str,
     entity_df_event_timestamp_col: str,
+    full_feature_names: bool = False,
 ):
     """Build point-in-time query between each feature view table and the entity dataframe"""
     template = Environment(loader=BaseLoader()).from_string(
@@ -448,6 +452,7 @@ def build_point_in_time_query(
             [entity for fv in feature_view_query_contexts for entity in fv.entities]
         ),
         "featureviews": [asdict(context) for context in feature_view_query_contexts],
+        "full_feature_names": full_feature_names,
     }
 
     query = template.render(template_context)
@@ -521,7 +526,7 @@ def _get_bigquery_client(project: Optional[str] = None):
         {{ featureview.created_timestamp_column ~ ' as created_timestamp,' if featureview.created_timestamp_column else '' }}
         {{ featureview.entity_selections | join(', ')}},
         {% for feature in featureview.features %}
-            {{ feature }} as {{ featureview.name }}__{{ feature }}{% if loop.last %}{% else %}, {% endif %}
+            {{ feature }} as {% if full_feature_names %}{{ featureview.name }}__{{feature}}{% else %}{{ feature }}{% endif %}{% if loop.last %}{% else %}, {% endif %}
         {% endfor %}
     FROM {{ featureview.table_subquery }}
 ),
@@ -614,7 +619,7 @@ def _get_bigquery_client(project: Optional[str] = None):
     SELECT
         entity_row_unique_id,
         {% for feature in featureview.features %}
-            {{ featureview.name }}__{{ feature }},
+            {% if full_feature_names %}{{ featureview.name }}__{{feature}}{% else %}{{ feature }}{% endif %},
         {% endfor %}
     FROM {{ featureview.name }}__cleaned
 ) USING (entity_row_unique_id)