[FSTORE-1672][4.1][APPEND] Allow multiple on-demand features to be returned from an on-demand transformation function and allow passing of local variables to a transformation function (logicalclocks#473)

* adding unknown type that can be used to used to initialize transformation types when a transformation function is not attached to a feature view or a fetaure group

* initalizing output column names only if  transformation type is not UNDEFINED

* handling undefined transformation type during feature view and feature grouo creation

* explicitly checking if features is a list or None to handle usecase where features is DataFrame

* adding backwards compatiblity to 4.1.x versions of hopsworks

* adding backwards compatiblity to 4.1.x versions of hopsworks

* fixing tests by mockign connections

* fixing unit tests

* handling versions with SNAPSHOT, updating comments and removing prints

* skipping test that requires Versions above 4.1.6

* addressing review comments

* setting transformation function types to be send to the backend only on versions above 4.1.6

* reverting changes

* fixing tests

Author: manu-sj

python/hopsworks_common/connection.py

@@ -142,6 +142,7 @@ def __init__(
         self._api_key_file = api_key_file
         self._api_key_value = api_key_value
         self._connected = False
+        self._backend_version = None
 
         self.connect()
 
@@ -299,16 +300,16 @@ def _check_compatibility(self):
         regexMatcher = re.compile(versionPattern)
 
         client_version = version.__version__
-        backend_version = self._variable_api.get_version("hopsworks")
+        self.backend_version = self._variable_api.get_version("hopsworks")
 
         major_minor_client = regexMatcher.search(client_version).group(0)
-        major_minor_backend = regexMatcher.search(backend_version).group(0)
+        major_minor_backend = regexMatcher.search(self._backend_version).group(0)
 
         if major_minor_backend != major_minor_client:
             print("\n", file=sys.stderr)
             warnings.warn(
                 "The installed hopsworks client version {0} may not be compatible with the connected Hopsworks backend version {1}. \nTo ensure compatibility please install the latest bug fix release matching the minor version of your backend ({2}) by running 'pip install hopsworks=={2}.*'".format(
-                    client_version, backend_version, major_minor_backend
+                    client_version, self._backend_version, major_minor_backend
                 ),
                 stacklevel=1,
             )
@@ -614,6 +615,23 @@ def api_key_file(self) -> Optional[str]:
     def api_key_value(self) -> Optional[str]:
         return self._api_key_value
 
+    @property
+    def backend_version(self) -> Optional[str]:
+        """
+        The version of the backend currently connected to hopsworks.
+        """
+        return self._backend_version
+
+    @backend_version.setter
+    def backend_version(self, backend_version: str) -> None:
+        """
+        The version of the backend currently connected to hopsworks.
+        """
+        self._backend_version = backend_version.split("-SNAPSHOT")[
+            0
+        ].strip()  # Strip off the -SNAPSHOT part of the version if it is present.
+        return self._backend_version
+
     @api_key_file.setter
     @not_connected
     def api_key_file(self, api_key_file: Optional[str]) -> None:

python/hsfs/feature_group.py

@@ -2379,7 +2379,11 @@ def __init__(
                         )
                     )
                 else:
-                    if not transformation_function.transformation_type:
+                    if (
+                        not transformation_function.transformation_type
+                        or transformation_function.transformation_type
+                        == TransformationType.UNDEFINED
+                    ):
                         transformation_function.transformation_type = (
                             TransformationType.ON_DEMAND
                         )
@@ -2730,11 +2734,11 @@ def save(
         if (
             (features is None and len(self._features) > 0)
             or (
-                isinstance(features, List)
+                isinstance(features, list)
                 and len(features) > 0
                 and all([isinstance(f, feature.Feature) for f in features])
             )
-            or (not features and len(self.transformation_functions) > 0)
+            or (features is None and len(self.transformation_functions) > 0)
         ):
             # This is done for compatibility. Users can specify the feature list in the
             # (get_or_)create_feature_group. Users can also provide the feature list in the save().

python/hsfs/feature_view.py

@@ -164,7 +164,11 @@ def __init__(
                         )
                     )
                 else:
-                    if not transformation_function.transformation_type:
+                    if (
+                        not transformation_function.transformation_type
+                        or transformation_function.transformation_type
+                        == TransformationType.UNDEFINED
+                    ):
                         transformation_function.transformation_type = (
                             TransformationType.MODEL_DEPENDENT
                         )

python/hsfs/hopsworks_udf.py

@@ -26,12 +26,14 @@
 from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
 
 import humps
+from hopsworks_common import client
 from hopsworks_common.client.exceptions import FeatureStoreException
 from hopsworks_common.constants import FEATURES
 from hsfs import engine, util
 from hsfs.core.feature_descriptive_statistics import FeatureDescriptiveStatistics
 from hsfs.decorators import typechecked
 from hsfs.transformation_statistics import TransformationStatistics
+from packaging.version import Version
 
 
 class UDFExecutionMode(Enum):
@@ -857,6 +859,8 @@ def to_dict(self) -> Dict[str, Any]:
         # Returns
             `Dict`: Dictionary that contains all data required to json serialize the object.
         """
+        backend_version = client.get_connection().backend_version
+
         return {
             "sourceCode": self._function_source,
             "outputTypes": self.return_types,
@@ -869,7 +873,11 @@ def to_dict(self) -> Dict[str, Any]:
             "name": self.function_name,
             "featureNamePrefix": self._feature_name_prefix,
             "executionMode": self.execution_mode.value.upper(),
-            "outputColumnNames": self.output_column_names,
+            **(
+                {"outputColumnNames": self.output_column_names}
+                if Version(backend_version) >= Version("4.1.6")
+                else {}
+            ),  # This check is added for backward compatibility with older versions of Hopsworks. The "outputColumnNames" field was added in Hopsworks 4.1.6 and versions below do not support unknown fields in the backend.
         }
 
     def json(self) -> str:

python/hsfs/transformation_function.py

@@ -21,6 +21,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 import humps
+from hopsworks_common import client
 from hopsworks_common.client.exceptions import FeatureStoreException
 from hopsworks_common.constants import FEATURES
 from hsfs import util
@@ -29,6 +30,7 @@
 from hsfs.decorators import typechecked
 from hsfs.hopsworks_udf import HopsworksUdf
 from hsfs.transformation_statistics import TransformationStatistics
+from packaging.version import Version
 
 
 _logger = logging.getLogger(__name__)
@@ -41,6 +43,7 @@ class TransformationType(Enum):
 
     MODEL_DEPENDENT = "model_dependent"
     ON_DEMAND = "on_demand"
+    UNDEFINED = "undefined"  # This type is used when the UDF created is not attached to a feature view / feature group. Hence the transformation function is neither model dependent nor on-demand.
 
 
 @typechecked
@@ -87,7 +90,12 @@ def __init__(
         TransformationFunction._validate_transformation_type(
             transformation_type=transformation_type, hopsworks_udf=hopsworks_udf
         )
-        self.transformation_type = transformation_type
+
+        # Setting transformation type as unknown when the transformation function is not attached to a feature view / feature group.
+        # This happens for example when the transformation function is fetched from the backend.
+        self.transformation_type = (
+            transformation_type if transformation_type else TransformationType.UNDEFINED
+        )
 
         if self.__hopsworks_udf._generate_output_col_name:
             # Reset output column names so that they would be regenerated.
@@ -230,12 +238,18 @@ def to_dict(self) -> Dict[str, Any]:
         # Returns
             `Dict`: Dictionary that contains all data required to json serialize the object.
         """
+        backend_version = client.get_connection().backend_version
+
         return {
             "id": self._id,
             "version": self._version,
             "featurestoreId": self._featurestore_id,
             "hopsworksUdf": self.hopsworks_udf.to_dict(),
-            "transformationType": self.transformation_type.value,
+            **(
+                {"transformationType": self.transformation_type.value}
+                if Version(backend_version) > Version("4.1.6")
+                else {}
+            ),  # This check is added for backward compatibility with older versions of Hopsworks. The "transformationType" field was added for equality checking of transformation functions and versions below 4.1.6 do not support unknown fields in the backend.
         }
 
     def alias(self, *args: str):
@@ -349,7 +363,11 @@ def version(self, version: int) -> None:
     def hopsworks_udf(self) -> HopsworksUdf:
         """Meta data class for the user defined transformation function."""
         # Make sure that the output column names for a model-dependent or on-demand transformation function, when accessed externally from the class.
-        if self.transformation_type and not self.__hopsworks_udf.output_column_names:
+        if (
+            self.transformation_type
+            and self.transformation_type != TransformationType.UNDEFINED
+            and not self.__hopsworks_udf.output_column_names
+        ):
             self.__hopsworks_udf.output_column_names = self._get_output_column_names()
         return self.__hopsworks_udf
 

python/tests/test_feature_view.py

@@ -15,6 +15,7 @@
 #
 import warnings
 
+from hopsworks_common import version
 from hsfs import feature_view, training_dataset_feature
 from hsfs.constructor import fs_query, query
 from hsfs.feature_store import FeatureStore
@@ -148,6 +149,11 @@ def test_from_response_json_basic_info_deprecated(self, mocker, backend_fixtures
     def test_transformation_function_instances(self, mocker, backend_fixtures):
         # Arrange
         feature_store_id = 99
+        mocked_connection = mocker.MagicMock()
+        mocked_connection.backend_version = version.__version__
+        mocked_connection = mocker.patch(
+            "hopsworks_common.client.get_connection", return_value=mocked_connection
+        )
         mocker.patch("hsfs.core.feature_view_engine.FeatureViewEngine")
         json = backend_fixtures["fs_query"]["get"]["response"]
 

python/tests/test_transformation_function.py

@@ -19,9 +19,11 @@
 
 import pandas as pd
 import pytest
+from hopsworks_common import version
 from hsfs.client.exceptions import FeatureStoreException
 from hsfs.hopsworks_udf import udf
 from hsfs.transformation_function import TransformationFunction, TransformationType
+from packaging.version import Version
 
 
 class TestTransformationFunction:
@@ -228,7 +230,15 @@ def test(col1):
             == "Please use the hopsworks_udf decorator when defining transformation functions."
         )
 
-    def test_transformation_function_definition_with_hopworks_udf(self):
+    def test_transformation_function_definition_with_hopworks_udf(self, mocker):
+        mocked_connection = mocker.MagicMock()
+        mocked_connection.backend_version = (
+            version.__version__
+        )  # Mocking backend version to be the same as the current version
+        mocked_connection = mocker.patch(
+            "hopsworks_common.client.get_connection", return_value=mocked_connection
+        )
+
         @udf(int)
         def test2(col1):
             return col1 + 1
@@ -956,7 +966,19 @@ def really_long_function_name_that_exceed_63_characters_causing_invalid_name_for
             "really_long_function_name_that_exceed_63_characters_causing_inv"
         ]
 
-    def test_equality_mdt(self):
+    @pytest.mark.skipif(
+        Version("4.1.6") >= Version(version.__version__),
+        reason="Requires Hopsworks 4.1.7 or higher to be working.",
+    )
+    def test_equality_mdt(self, mocker):
+        mocked_connection = mocker.MagicMock()
+        mocked_connection.backend_version = (
+            version.__version__
+        )  # Mocking backend version to be the same as the current version
+        mocked_connection = mocker.patch(
+            "hopsworks_common.client.get_connection", return_value=mocked_connection
+        )
+
         @udf([int])
         def add_one(feature):
             return feature + 1
@@ -975,7 +997,19 @@ def add_one(feature):
 
         assert mdt1 == mdt2
 
-    def test_equality_odt(self):
+    @pytest.mark.skipif(
+        Version("4.1.6") >= Version(version.__version__),
+        reason="Requires Hopsworks 4.1.7 or higher to be working.",
+    )
+    def test_equality_odt(self, mocker):
+        mocked_connection = mocker.MagicMock()
+        mocked_connection.backend_version = (
+            version.__version__
+        )  # Mocking backend version to be the same as the current version
+        mocked_connection = mocker.patch(
+            "hopsworks_common.client.get_connection", return_value=mocked_connection
+        )
+
         @udf([int])
         def add_one(feature):
             return feature + 1
@@ -994,7 +1028,19 @@ def add_one(feature):
 
         assert odt1 == odt2
 
-    def test_inequality(self):
+    @pytest.mark.skipif(
+        Version("4.1.6") >= Version(version.__version__),
+        reason="Requires Hopsworks 4.1.7 or higher to be working.",
+    )
+    def test_inequality(self, mocker):
+        mocked_connection = mocker.MagicMock()
+        mocked_connection.backend_version = (
+            version.__version__
+        )  # Mocking backend version to be the same as the current version
+        mocked_connection = mocker.patch(
+            "hopsworks_common.client.get_connection", return_value=mocked_connection
+        )
+
         @udf([int])
         def add_one(feature):
             return feature + 1