[FSTORE-1411][APPEND] On-Demand Transformations (#340)

* using deep copy to copy feature group object instead of from_response

* reverting deep copy change and using to_dict instead

* using get to check if the attributes present in response dict

Author: manu-sj

python/hsfs/core/feature_group_engine.py

@@ -259,7 +259,7 @@ def sql(self, query, feature_store_name, dataframe_type, online, read_options):
             read_options,
         )
 
-    def _update_features_metadata(self, feature_group, features):
+    def _update_features_metadata(self, feature_group: fg.FeatureGroup, features):
         # perform changes on copy in case the update fails, so we don't leave
         # the user object in corrupted state
         copy_feature_group = fg.FeatureGroup.from_response_json(feature_group.to_dict())

python/hsfs/feature_group.py

@@ -3554,7 +3554,9 @@ def to_dict(self) -> Dict[str, Any]:
             "topicName": self.topic_name,
             "notificationTopicName": self.notification_topic_name,
             "deprecated": self.deprecated,
-            "transformationFunctions": self._transformation_functions,
+            "transformationFunctions": [
+                tf.to_dict() for tf in self._transformation_functions
+            ],
         }
         if self._online_config:
             fg_meta_dict["onlineConfig"] = self._online_config.to_dict()

python/hsfs/hopsworks_udf.py

@@ -315,7 +315,7 @@ def _extract_source_code(udf_function: Callable) -> str:
         except FileNotFoundError:
             module_imports = [""]
             warnings.warn(
-                "Cannot extract imported dependencies for the function module. Please make sure to import all dependencies for the UDF inside the function.",
+                "Cannot extract imported dependencies for the UDF from the module in which it is defined. Please make sure to import all dependencies for the UDF inside the function.",
                 stacklevel=2,
             )
 
@@ -657,7 +657,7 @@ def from_response_json(
                 dropped_feature.strip()
                 for dropped_feature in json_decamelized["dropped_argument_names"]
             ]
-            if "dropped_argument_names" in json_decamelized
+            if json_decamelized.get("dropped_argument_names", None)
             else None
         )
         transformation_function_argument_names = (
@@ -667,15 +667,15 @@ def from_response_json(
                     "transformation_function_argument_names"
                 ]
             ]
-            if "transformation_function_argument_names" in json_decamelized
+            if json_decamelized.get("transformation_function_argument_names", None)
             else None
         )
         statistics_features = (
             [
                 feature.strip()
                 for feature in json_decamelized["statistics_argument_names"]
             ]
-            if "statistics_argument_names" in json_decamelized
+            if json_decamelized.get("statistics_argument_names", None)
             else None
         )
 

python/hsfs/transformation_function.py

@@ -227,7 +227,7 @@ def to_dict(self) -> Dict[str, Any]:
             "id": self._id,
             "version": self._version,
             "featurestoreId": self._featurestore_id,
-            "hopsworksUdf": self._hopsworks_udf,
+            "hopsworksUdf": self._hopsworks_udf.to_dict(),
         }
 
     def _get_output_column_names(self) -> str: