adding soft fail for dataframes, and adding extended type

Author: manu-sj

python/hopsworks_common/core/type_systems.py

@@ -19,7 +19,7 @@
 import ast
 import datetime
 import decimal
-from typing import TYPE_CHECKING, Literal, Union
+from typing import TYPE_CHECKING, Literal, NewType, Union
 
 import pytz
 from hopsworks_common.core.constants import (
@@ -132,6 +132,37 @@
     }
 
 
+def create_extended_type(base_type: type) -> "ExtendedType":
+    """
+    This is wrapper function to create a new class that extends the base_type class with a new attribute that can be used to store metadata.
+
+    Args:
+        base_type : The base class to extend
+    """
+
+    class ExtendedType(base_type):
+        """
+        This is a class that extends the base_type class with a new attribute `hopsworks_meta_data` that can be used to store metadata.
+        """
+
+        @property
+        def hopsworks_meta_data(self):
+            if not hasattr(self, "_hopsworks_meta_data"):
+                return None
+            return self._hopsworks_meta_data
+
+        @hopsworks_meta_data.setter
+        def hopsworks_meta_data(self, meta_data: dict):
+            self._hopsworks_meta_data = meta_data
+
+    return ExtendedType
+
+
+ExtendedType = NewType(
+    "ExtendedType", create_extended_type(type)
+)  # Adding new type for type hinting and static analysis.
+
+
 def convert_pandas_dtype_to_offline_type(arrow_type: str) -> str:
     # This is a simple type conversion between pandas dtypes and pyspark (hive) types,
     # using pyarrow types obatined from pandas dataframe to convert pandas typed fields,

python/hsfs/core/feature_view_engine.py

@@ -57,7 +57,7 @@ class FeatureViewEngine:
 
     _LOG_TD_VERSION = "td_version"
     _LOG_TIME = "log_time"
-    _HSML_MODEL = "hsml_model"
+    _HSML_MODEL = "hopsworks_model"
 
     def __init__(self, feature_store_id):
         self._feature_store_id = feature_store_id

python/hsfs/core/vector_server.py

@@ -1308,8 +1308,13 @@ def apply_transformation(
     ):
         """
         Function that applies both on-demand and model dependent transformation to the input dictonary
+
+        Returns:
+            feature_vector: The untransformed feature vector with untransformed features.
+            encoded_feature_dict: The transformed feature vector with transformed features.
         """
-        encoded_feature_dict = row_dict
+        feature_dict = row_dict
+        encoded_feature_dict = None
 
         if transform or on_demand_features:
             # Check for any missing request parameters
@@ -1318,17 +1323,17 @@ def apply_transformation(
             )
 
             # Apply on-demand transformations
-            encoded_feature_dict = self.apply_on_demand_transformations(
+            feature_dict = self.apply_on_demand_transformations(
                 row_dict, request_parameter, transformation_context
             )
 
         if transform:
             # Apply model dependent transformations
             encoded_feature_dict = self.apply_model_dependent_transformations(
-                encoded_feature_dict, transformation_context
+                feature_dict, transformation_context
             )
 
-        return encoded_feature_dict
+        return feature_dict, encoded_feature_dict
 
     def apply_return_value_handlers(
         self, row_dict: Dict[str, Any], client: Literal["rest", "sql"]

python/hsfs/engine/python.py

@@ -1707,6 +1707,8 @@ def get_feature_logging_df(
                 predictions[f.name] = cast_column_to_offline_type(
                     predictions[f.name], f.type
                 )
+            # Addin the prefix prediction_ to the prediction columns
+            predictions = predictions.add_prefix("predicted_")
             if not set(predictions.columns).intersection(set(features.columns)):
                 features = pd.concat([features, predictions], axis=1)
 
@@ -1723,6 +1725,28 @@ def get_feature_logging_df(
             features[k] = pd.Series(v)
         # _cast_column_to_offline_type cannot cast string type
         features[model_col_name] = features[model_col_name].astype(pd.StringDtype())
+
+        logging_feature_group_features = [feat.name for feat in fg.features]
+        missing_logging_features = set(logging_feature_group_features).difference(
+            set(features.columns)
+        )
+        additional_logging_features = set(features.columns).difference(
+            set(logging_feature_group_features)
+        )
+
+        if additional_logging_features:
+            _logger.info(
+                f"The following columns : `{'`, `'.join(additional_logging_features)}` are additional columns in the logged {'untransformed' if 'untransformed' in fg.name else 'transformed'} dataframe and is not present in the logging feature groups. They will be ignored."
+            )
+
+        if missing_logging_features:
+            _logger.info(
+                f"The following columns : `{'`, `'.join(missing_logging_features)}` are missing in the logged {'untransformed' if 'untransformed' in fg.name else 'transformed'} dataframe. Setting them to None."
+            )
+            # Set missing columns to None
+            for col in missing_logging_features:
+                features[col] = None
+
         return features[[feat.name for feat in fg.features]]
 
     @staticmethod