fixed the query builder for joins

Author: yogesh266

ads/feature_store/execution_strategy/spark/spark_execution.py

@@ -179,7 +179,9 @@ def _save_offline_dataframe(
 
             if isinstance(data_frame, pd.DataFrame):
                 if not feature_group.is_infer_schema:
-                    convert_pandas_datatype_with_schema(feature_group.input_feature_details, data_frame)
+                    convert_pandas_datatype_with_schema(
+                        feature_group.input_feature_details, data_frame
+                    )
 
             # TODO: Get event timestamp column and apply filtering basis from and to timestamp
 

ads/feature_store/feature_group.py

@@ -155,7 +155,7 @@ class FeatureGroup(Builder):
         CONST_LIFECYCLE_STATE: "lifecycle_state",
         CONST_OUTPUT_FEATURE_DETAILS: "output_feature_details",
         CONST_STATISTICS_CONFIG: "statistics_config",
-        CONST_INFER_SCHEMA: "is_infer_schema"
+        CONST_INFER_SCHEMA: "is_infer_schema",
     }
 
     def __init__(self, spec: Dict = None, **kwargs) -> None:
@@ -1225,7 +1225,8 @@ def _get_job_id(self, job_id: str = None) -> str:
 
         if self.job_id is None:
             raise ValueError(
-                "Unable to retrieve the last job. Please provide the job ID and make sure you materialized the data.")
+                "Unable to retrieve the last job. Please provide the job ID and make sure you materialized the data."
+            )
 
         return self.job_id
 
@@ -1278,9 +1279,13 @@ def get_validation_output(self, job_id: str = None) -> "ValidationOutput":
         # Retrieve the validation output JSON from data_flow_batch_execution_output.
         fg_job = FeatureGroupJob.from_id(validation_job_id)
         output_details = fg_job.job_output_details
-        validation_output = output_details.get("validationOutput") if output_details else None
+        validation_output = (
+            output_details.get("validationOutput") if output_details else None
+        )
 
-        validation_output_json = json.loads(validation_output) if validation_output else None
+        validation_output_json = (
+            json.loads(validation_output) if validation_output else None
+        )
 
         return ValidationOutput(validation_output_json)
 

ads/feature_store/query/generator/query_generator.py

@@ -74,12 +74,13 @@ def generate_query(self, is_online: bool = False) -> str:
         selected_features_map = {}
         index = 0
         on_condition = []
-        left_table = f"`{self.query.entity_id}`.{self.query.left_feature_group.name}"
-        left_table_alias = self.get_table_alias(len(self.query.joins))
+        table = f"`{self.query.entity_id}`.{self.query.left_feature_group.name}"
+        table_alias = self.get_table_alias(len(self.query.joins))
+        left_table_alias = table_alias
 
         # store the left features in the map
-        selected_features_map[left_table_alias] = self.query.left_features
-        feature_group_id_map = {self.query.left_feature_group.id: left_table_alias}
+        selected_features_map[table_alias] = self.query.left_features
+        feature_group_id_map = {self.query.left_feature_group.id: table_alias}
 
         for join in self.query.joins:
             # Ge table and alias and map the features
@@ -99,6 +100,8 @@ def generate_query(self, is_online: bool = False) -> str:
                 )
             )
 
+            left_table_alias = right_table_alias
+
         selected_columns = self._get_selected_columns(selected_features_map)
         filters = (
             None
@@ -109,7 +112,7 @@ def generate_query(self, is_online: bool = False) -> str:
         )
 
         # Define the SQL query as an f-string
-        query = f"SELECT {selected_columns} FROM {left_table} {left_table_alias}"
+        query = f"SELECT {selected_columns} FROM {table} {table_alias}"
 
         if on_condition:
             # If there is an ON condition, add it to the query

ads/feature_store/query/query.py

@@ -109,6 +109,20 @@ def with_left_feature_group(self, feature_group) -> "Query":
         """
         return self.set_spec(self.CONST_LEFT_FEATURE_GROUP, feature_group)
 
+    def get_last_joined_feature_group(self):
+        """
+        Retrieves the last joined feature group from the list of joins,
+        or returns the left feature group if no joins are present.
+
+        Returns:
+            The last joined feature group if the list of joins is non-empty,
+            otherwise returns the left feature group.
+        """
+        if self.joins:
+            return self.joins[-1].sub_query.left_feature_group
+        else:
+            return self.left_feature_group
+
     @property
     def _filter(self):
         return self.get_spec(self.CONST_FILTER)
@@ -291,7 +305,7 @@ def join(
         """
 
         join = Join(sub_query, on, left_on, right_on, join_type)
-        QueryValidator.validate_query_join(self.left_feature_group, join)
+        QueryValidator.validate_query_join(self.get_last_joined_feature_group(), join)
         self.joins.append(join)
         return self
 

ads/feature_store/response/response_builder.py

@@ -51,11 +51,9 @@ def to_pandas(self) -> pd.DataFrame:
 
     @property
     def kind(self) -> str:
-
         return "response_builder"
 
     def to_dict(self) -> Dict:
-
         spec = deepcopy(self._spec)
         for key, value in spec.items():
             if hasattr(value, "to_dict"):

ads/feature_store/statistics.py

@@ -22,4 +22,4 @@ def kind(self) -> str:
         str
             The kind of the statistics object, which is always "statistics".
         """
-        return "statistics"
+        return "statistics"