update to handle explicit features

Author: dhananjay-mk

python/hsfs/core/schema_validation.py

@@ -54,7 +54,7 @@ def validate_schema(self, feature_group, df, df_features):
                 )
         # Execute data type specific validation
         errors, column_lengths, is_pk_null, is_string_length_exceeded = (
-            self._validate_df_specifics(feature_group, df, bool(feature_group.id))
+            self._validate_df_specifics(feature_group, df)
         )
 
         # Handle errors
@@ -68,7 +68,7 @@ def validate_schema(self, feature_group, df, df_features):
 
         return df_features
 
-    def _validate_df_specifics(self, feature_group, df, is_fg_created):
+    def _validate_df_specifics(self, feature_group, df):
         """To be implemented by subclasses"""
         raise NotImplementedError("Subclasses must implement this method")
 
@@ -77,7 +77,8 @@ def get_feature_from_list(feature_name, features):
         for i_feature in features:
             if i_feature.name == feature_name:
                 return i_feature
-        raise ValueError(f"Feature {feature_name} not found in feature list")
+
+        return None
 
     @staticmethod
     def extract_numbers(input_string):
@@ -87,13 +88,14 @@ def extract_numbers(input_string):
         return re.findall(pattern, input_string)
 
     def get_online_varchar_length(self, feature):
-        # returns the column length of varchar columns
-        if not feature.type == "string":
-            raise ValueError("Feature not a string type")
-        if not feature.online_type:
-            raise ValueError("Feature is not online enabled")
-
-        return int(self.extract_numbers(feature.online_type)[0])
+        # check of online_type is not null and starts with varchar
+        if (
+            feature
+            and feature.online_type
+            and feature.online_type.startswith("varchar")
+        ):
+            return int(self.extract_numbers(feature.online_type)[0])
+        return None
 
     @staticmethod
     def increase_string_columns(column_lengths: dict, dataframe_features):
@@ -109,7 +111,7 @@ def increase_string_columns(column_lengths: dict, dataframe_features):
 
 class PandasValidator(DataFrameValidator):
     # Pandas df specific validator
-    def _validate_df_specifics(self, feature_group, df, is_fg_created):
+    def _validate_df_specifics(self, feature_group, df):
         errors = {}
         column_lengths = {}
         is_pk_null = False
@@ -118,7 +120,7 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
         # Check for null values in primary key columns
         for pk in feature_group.primary_key:
             if df[pk].isnull().any():
-                errors[pk] = f"Primary key column {pk} contains null values"
+                errors[pk] = f"Primary key column {pk} contains null values."
                 is_pk_null = True
 
         # Check string lengths
@@ -128,13 +130,13 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
                 self.get_online_varchar_length(
                     self.get_feature_from_list(col, feature_group.features)
                 )
-                if is_fg_created
+                if feature_group.features
                 else 100
             )
 
-            if currentmax > col_max_len:
+            if col_max_len is not None and currentmax > col_max_len:
                 errors[col] = (
-                    f"Column {col} has string values longer than {col_max_len} characters"
+                    f"String length exceeded. Column {col} has string values longer than maximum colum limit of {col_max_len} characters."
                 )
                 column_lengths[col] = currentmax
                 is_string_length_exceeded = True
@@ -144,7 +146,7 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
 
 class PolarsValidator(DataFrameValidator):
     # Polars df specific validator
-    def _validate_df_specifics(self, feature_group, df, is_fg_created):
+    def _validate_df_specifics(self, feature_group, df):
         import polars as pl
 
         errors = {}
@@ -155,7 +157,7 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
         # Check for null values in primary key columns
         for pk in feature_group.primary_key:
             if df[pk].is_null().any():
-                errors[pk] = f"Primary key column {pk} contains null values"
+                errors[pk] = f"Primary key column {pk} contains null values."
                 is_pk_null = True
 
         # Check string lengths
@@ -165,13 +167,13 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
                 self.get_online_varchar_length(
                     self.get_feature_from_list(col, feature_group.features)
                 )
-                if is_fg_created
+                if feature_group.features
                 else 100
             )
 
-            if currentmax > col_max_len:
+            if col_max_len is not None and currentmax > col_max_len:
                 errors[col] = (
-                    f"Column {col} has string values longer than {col_max_len} characters"
+                    f"String length exceeded. Column {col} has string values longer than maximum colum limit of {col_max_len} characters."
                 )
                 column_lengths[col] = currentmax
                 is_string_length_exceeded = True
@@ -181,7 +183,7 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
 
 class PySparkValidator(DataFrameValidator):
     # PySpark-specific validator
-    def _validate_df_specifics(self, feature_group, df, is_fg_created):
+    def _validate_df_specifics(self, feature_group, df):
         # Import PySpark SQL functions and types
         import pyspark.sql.functions as sf
         from pyspark.sql.types import StringType
@@ -194,7 +196,7 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
         # Check for null values in primary key columns
         for pk in feature_group.primary_key:
             if df.filter(df[pk].isNull()).count() > 0:
-                errors[pk] = f"Primary key column {pk} contains null values"
+                errors[pk] = f"Primary key column {pk} contains null values."
                 is_pk_null = True
 
         # Check string lengths for string columns
@@ -209,13 +211,13 @@ def _validate_df_specifics(self, feature_group, df, is_fg_created):
                     self.get_online_varchar_length(
                         self.get_feature_from_list(col, feature_group.features)
                     )
-                    if is_fg_created
+                    if feature_group.features
                     else 100
                 )
 
-                if currentmax > col_max_len:
+                if col_max_len is not None and currentmax > col_max_len:
                     errors[col] = (
-                        f"Column {col} has string values longer than {col_max_len} characters"
+                        f"String length exceeded. Column {col} has string values longer than maximum colum limit of {col_max_len} characters."
                     )
                     column_lengths[col] = currentmax
                     is_string_length_exceeded = True

python/tests/test_schema_validator.py

@@ -28,7 +28,7 @@ def pandas_df():
                         for _ in range(random.randint(1, 100))
                     )
                 )
-                for i in range(3)
+                for _ in range(3)
             ],
         }
     )
@@ -47,7 +47,7 @@ def polars_df():
                         for _ in range(random.randint(1, 100))
                     )
                 )
-                for i in range(3)
+                for _ in range(3)
             ],
         }
     )
@@ -138,9 +138,7 @@ def test_validate_schema_string_length_exceeded(
         self, pandas_df, feature_group_created, mocker
     ):
         pandas_df.loc[0, "string_col"] = "a" * 101
-        with pytest.raises(
-            ValueError, match="Column string_col has string values longer than 100"
-        ):
+        with pytest.raises(ValueError, match="String length exceeded"):
             DataFrameValidator().validate_schema(
                 feature_group_created, pandas_df, feature_group_created.features
             )
@@ -149,22 +147,38 @@ def test_validate_schema_feature_group_created(
         self, pandas_df, feature_group_created, mocker
     ):
         pandas_df.loc[0, "string_col"] = "a" * 101
-        with pytest.raises(
-            ValueError, match="Column string_col has string values longer"
-        ):
+        with pytest.raises(ValueError, match="String length exceeded"):
             DataFrameValidator().validate_schema(
                 feature_group_created, pandas_df, feature_group_created.features
             )
 
     def test_validate_schema_feature_group_not_created(
         self, pandas_df, feature_group_data
     ):
+        # test with non existing feature group with no explicit features
+        # arrange
         pandas_df.loc[0, "string_col"] = "a" * 101
+        initial_features = [
+            Feature("primary_key", "int"),
+            Feature("event_time", "string"),
+            Feature("string_col", "string"),
+        ]
+        feature_group_data.features = []
         df_features = DataFrameValidator().validate_schema(
-            feature_group_data, pandas_df, feature_group_data.features
+            feature_group_data, pandas_df, initial_features
         )
         assert df_features[2].online_type == "varchar(101)"
 
+    def test_validate_schema_feature_group_with_features_not_created(
+        self, pandas_df, feature_group_data
+    ):
+        # test with feature group with explicit features
+        df_features = DataFrameValidator().validate_schema(
+            feature_group_data, pandas_df, feature_group_data.features
+        )
+        # assert that the online type of the string_col feature is same as explcitly set in the feature group
+        assert df_features[2].online_type == "varchar(200)"
+
     def test_pk_null_string_length_exceeded(self, pandas_df, feature_group_data):
         pandas_df.loc[0, "primary_key"] = None
         pandas_df.loc[0, "string_col"] = "a" * 101
@@ -180,5 +194,18 @@ def test_offline_fg(self, pandas_df, feature_group_data, caplog):
         df_features = PandasValidator().validate_schema(
             feature_group_data, pandas_df, feature_group_data.features
         )
+        # assert no changes were made
+        assert df_features == feature_group_data.features
+
+    def test_should_not_update_nonvarchar(self, pandas_df, feature_group_data):
+        # Test that the validator does not update the online type of a non-varchar column
+        # arrange
+        # set string_col feature online type to text
+        feature_group_data.features[2].online_type = "text"
+        pandas_df.loc[0, "string_col"] = "b" * 1001
+        # act
+        df_features = PandasValidator().validate_schema(
+            feature_group_data, pandas_df, feature_group_data.features
+        )
+
         assert df_features == feature_group_data.features
-        assert "Feature group is not online enabled. Skipping validation" in caplog.text