refactoring changes

Author: manu-sj

python/hsfs/core/feature_view_engine.py

@@ -32,6 +32,8 @@
 )
 from hsfs.client import exceptions
 from hsfs.constructor.filter import Filter, Logic
+from hsfs.constructor.join import Join
+from hsfs.constructor.query import Query
 from hsfs.core import (
     feature_view_api,
     query_constructor_api,
@@ -1073,116 +1075,156 @@ def _get_feature_view_url(self, fv: "feature_view.FeatureView"):
     def generate_fully_qualied_feature_name(
         self, feature_group: feature_group.FeatureGroup, feature_name: str
     ):
+        """
+        Generate the fully qualified feature name for a feature. The fully qualified name is created by concatenating
+        the project name, feature group name, feature group version and feature name.
+        """
         return f"{feature_group._get_project_name()}_{feature_group.name}_{feature_group.version}_{feature_name}"
 
-    def _primary_keys_from_join(self, joins, check_duplicate):
-        query_pks = set()
+    def _primary_keys_from_join(
+        self, joins: List[Join], check_duplicate: bool, pk_names: set[str]
+    ) -> set[str]:
+        """
+        Recursive function that extracts the primary keys from the join objects and returns them as a set.
+        """
         for join in joins:
-            query = join.query
-            select_features = [feature.name for feature in join.query._left_features]
-            for feature in query._left_feature_group.features:
-                if feature.primary:
-                    if feature.name not in select_features:
-                        f_name = self.generate_fully_qualied_feature_name(
-                            query._left_feature_group, feature.name
-                        )
-                        if check_duplicate:
-                            # TODO: THis is not correct we must check the entire feature views selected features for duplicates need to adapt.
-                            f_name = self._check_if_exists_with_prefix(
-                                f_name, query_pks
-                            )
-                        query_pks.add(f_name)
-                    else:
-                        f_name = (
-                            feature.name
-                            if join.prefix is None
-                            else join.prefix + feature.name
-                        )
-                        if check_duplicate:
-                            # TODO: THis is not correct we must check the entire feature views selected features for duplicates need to adapt.
-                            f_name = self._check_if_exists_with_prefix(
-                                f_name, query_pks
-                            )
-                    query_pks.add(f_name)
-            if query._joins:
+            sub_query = join.query
+            join_prefix = join.prefix
+
+            sub_query_selected_feature_names = [
+                feature.name for feature in join.query._left_features
+            ]
+            sub_query_feature_group = sub_query._left_feature_group
+            sub_query_pk_names = {
+                feature.name
+                for feature in sub_query_feature_group.features
+                if feature.primary
+            }
+
+            sub_query_pk_names = {
+                self.generate_fully_qualied_feature_name(
+                    sub_query_feature_group, pk_name
+                )
+                if pk_name not in sub_query_selected_feature_names
+                else (pk_name if not join_prefix else join_prefix + pk_name)
+                for pk_name in sub_query_pk_names
+            }
+
+            if check_duplicate:
+                for sub_query_pk_name in sub_query_pk_names:
+                    self._check_if_exists_with_prefix(sub_query_pk_name, pk_names)
+
+            pk_names.update(sub_query_pk_names)
+
+            if sub_query._joins:
                 sub_query_pks = self._primary_keys_from_join(
-                    query._joins, check_duplicate
+                    sub_query._joins, check_duplicate, pk_names
                 )
-                query_pks.update(sub_query_pks)
-        return query_pks
+                pk_names.update(sub_query_pks)
+        return pk_names
 
-    def _event_time_from_join(self, joins, check_duplicate):
-        query_ets = set()
+    def _event_time_from_join(
+        self, joins: List[Join], check_duplicate: bool, et_names: set[str]
+    ) -> set[str]:
         for join in joins:
-            query = join.query
-            select_features = [feature.name for feature in join.query._left_features]
-            if join.query._left_feature_group.event_time:
-                if join.query._left_feature_group.event_time not in select_features:
-                    f_name = self.generate_fully_qualied_feature_name(
-                        join.query._left_feature_group,
-                        join.query._left_feature_group.event_time,
-                    )
-                    if check_duplicate:
-                        # TODO: THis is not correct we must check the entire feature views selected features for duplicates need to adapt.
-                        f_name = self._check_if_exists_with_prefix(f_name, query_ets)
-                    query_ets.add(f_name)
-                else:
-                    f_name = (
-                        join.query._left_feature_group.event_time
-                        if join.prefix is None
-                        else join.prefix + join.query._left_feature_group.event_time
-                    )
-                    if check_duplicate:
-                        # TODO: THis is not correct we must check the entire feature views selected features for duplicates need to adapt.
-                        f_name = self._check_if_exists_with_prefix(f_name, query_ets)
-                    query_ets.add(f_name)
-            if query._joins:
-                sub_query_ets = self._event_time_from_join(
-                    query._joins, check_duplicate
-                )
-                query_ets.update(sub_query_ets)
-        return query_ets
-
-    def _get_primary_keys_from_query(self, fv_query_obj, check_duplicate=True):
-        # Patch for now to check there is a need to use the fully qualified name
-        select_features = [feature.name for feature in fv_query_obj._left_features]
-        fv_pks = set(
-            [
+            sub_query = join.query
+            join_prefix = join.prefix
+
+            sub_query_selected_feature_names = [
+                feature.name for feature in join.query._left_features
+            ]
+            sub_query_feature_group = sub_query._left_feature_group
+            sub_query_event_time = sub_query_feature_group.event_time
+
+            sub_query_event_time = (
                 self.generate_fully_qualied_feature_name(
-                    fv_query_obj._left_feature_group, feature.name
+                    sub_query_feature_group, sub_query_event_time
                 )
-                if feature.name not in select_features
-                else feature.name
-                for feature in fv_query_obj._left_feature_group.features
-                if feature.primary
-            ]
-        )
-        fv_pks.update(
-            self._primary_keys_from_join(fv_query_obj._joins, check_duplicate)
-        )
+                if sub_query_event_time not in sub_query_selected_feature_names
+                else (
+                    join_prefix + sub_query_event_time
+                    if join_prefix
+                    else sub_query_event_time
+                )
+            )
 
-        return list(fv_pks)
+            if check_duplicate:
+                self._check_if_exists_with_prefix(sub_query_event_time, et_names)
 
-    def _get_eventtimes_from_query(self, fv_query_obj, check_duplicate=True):
-        fv_events = set()
-        select_features = [feature.name for feature in fv_query_obj._left_features]
+            et_names.add(sub_query_event_time)
 
-        if fv_query_obj._left_feature_group.event_time:
-            if fv_query_obj._left_feature_group.event_time not in select_features:
-                fv_events.add(
-                    self.generate_fully_qualied_feature_name(
-                        fv_query_obj._left_feature_group,
-                        fv_query_obj._left_feature_group.event_time,
-                    )
+            if sub_query._joins:
+                sub_query_ets = self._event_time_from_join(
+                    sub_query._joins, check_duplicate, et_names
                 )
-            else:
-                fv_events.add(fv_query_obj._left_feature_group.event_time)
+                et_names.update(sub_query_ets)
+        return et_names
+
+    def _get_primary_keys_from_query(
+        self, query: Query, check_duplicate: bool = True
+    ) -> List[str]:
+        """
+        Function that checks the primary keys from the query object and returns them as a list.
+
+        #Arguments:
+            fv_query_obj : `Query`. Query object from which the primary keys are extracted.
+            check_duplicate : `bool`. Flag to check if the primary keys are duplicated in the query.
+        """
+        root_feature_group_selected_features_name = {
+            feature.name for feature in query._left_features
+        }
+
+        root_feature_group = query._left_feature_group
+
+        root_feature_group_primary_keys_names = {
+            feature.name for feature in root_feature_group.features if feature.primary
+        }
+
+        pk_names = {
+            self.generate_fully_qualied_feature_name(root_feature_group, pk_name)
+            if pk_name not in root_feature_group_selected_features_name
+            else pk_name
+            for pk_name in root_feature_group_primary_keys_names
+        }
+
+        pk_names.update(
+            self._primary_keys_from_join(query._joins, check_duplicate, pk_names)
+        )
+
+        return list(pk_names)
+
+    def _get_eventtimes_from_query(
+        self, query: Query, check_duplicate: bool = True
+    ) -> List[str]:
+        """
+        Function that checks the event times from the query object and returns them as a list.
+
+        #Arguments:
+            fv_query_obj : `Query`. Query object from which the event times are extracted.
+            check_duplicate : `bool`. Flag to check if the event times are duplicated in the query.
+        """
+        root_feature_group_selected_features_name = {
+            feature.name for feature in query._left_features
+        }
+
+        root_feature_group = query._left_feature_group
+
+        root_feature_group_event_time = root_feature_group.event_time
+
+        et_names = {
+            self.generate_fully_qualied_feature_name(
+                root_feature_group, root_feature_group_event_time
+            )
+            if root_feature_group_event_time
+            not in root_feature_group_selected_features_name
+            else root_feature_group_event_time
+        }
 
-        fv_events.update(
-            self._event_time_from_join(fv_query_obj._joins, check_duplicate)
+        et_names.update(
+            self._event_time_from_join(query._joins, check_duplicate, et_names)
         )
 
-        return list(fv_events)
+        return list(et_names)
 
     def _check_if_exists_with_prefix(self, f_name, f_set):
         if f_name in f_set:

python/hsfs/engine/python.py

@@ -609,15 +609,15 @@ def profile(
             else:
                 print(
                     "Data type could not be inferred for column '"
-                    + col
+                    + col.split(".")[-1]
                     + "'. Defaulting to 'String'",
                     file=sys.stderr,
                 )
                 dataType = "String"
 
             stat = self._convert_pandas_statistics(stats[col], dataType)
             stat["isDataTypeInferred"] = "false"
-            stat["column"] = col
+            stat["column"] = col.split(".")[-1]
             stat["completeness"] = 1
 
             final_stats.append(stat)

python/hsfs/feature.py

@@ -277,7 +277,7 @@ def __str__(self) -> str:
         return self.json()
 
     def __repr__(self) -> str:
-        return f"Feature({self._name!r}, {self._type!r}, {self._description!r}, {self._primary}, {self._foreign}, {self._partition}, {self._online_type!r}, {self._default_value!r}, {self._feature_group_id!r}, use_fully_qualified_name={self._use_fully_qualified_name})"
+        return f"Feature({self._name!r}, {self._type!r}, {self._description!r}, {self._primary}, {self._foreign}, {self._partition}, {self._online_type!r}, {self._default_value!r}, {self._feature_group_id!r})"
 
     def __hash__(self) -> int:
         return hash(f"{self.feature_group_id}_{self.name}")