update to append_features

Author: bubriks

python/hsfs/core/feature_group_engine.py

@@ -297,9 +297,9 @@ def append_features(self, feature_group, new_features):
 
         # write empty dataframe to update parquet schema
         if feature_group.time_travel_format == "DELTA":
-            engine.get_instance().add_cols_to_delta_table(feature_group, new_features)
+            engine.get_instance().add_cols_to_delta_table(feature_group)
         else:
-            engine.get_instance().save_empty_dataframe(feature_group, new_features=new_features)
+            engine.get_instance().save_empty_dataframe(feature_group)
 
     def update_description(self, feature_group, description):
         """Updates the description of a feature group."""

python/hsfs/engine/python.py

@@ -1212,13 +1212,11 @@ def save_stream_dataframe(
             "Stream ingestion is not available on Python environments, because it requires Spark as engine."
         )
 
-    def save_empty_dataframe(
-        self, feature_group: Union[FeatureGroup, ExternalFeatureGroup], new_features=None
-    ) -> None:
+    def save_empty_dataframe(self, feature_group: Union[FeatureGroup, ExternalFeatureGroup]) -> None:
         """Wrapper around save_dataframe in order to provide no-op."""
         pass
 
-    def add_cols_to_delta_table(self, feature_group: FeatureGroup, new_features) -> None:
+    def add_cols_to_delta_table(self, feature_group: FeatureGroup) -> None:
         """Wrapper around add_cols_to_delta_table in order to provide no-op."""
         pass
 

python/hsfs/engine/spark.py

@@ -1324,18 +1324,14 @@ def is_spark_dataframe(self, dataframe):
             return True
         return False
 
-    def save_empty_dataframe(self, feature_group, new_features=None):
+    def save_empty_dataframe(self, feature_group):
         location = feature_group.prepare_spark_location()
 
         dataframe = self._spark_session.read.format("hudi").load(location)
 
-        if (new_features is not None):
-            if isinstance(new_features, list):
-                for new_feature in new_features:
-                    dataframe = dataframe.withColumn(new_feature.name, lit(None).cast(new_feature.type))
-            else:
-                dataframe = dataframe.withColumn(new_features.name, lit(None).cast(new_features.type))
-
+        for feature in feature_group.features:
+            if feature.name not in dataframe.columns:
+                dataframe = dataframe.withColumn(feature.name, lit(None).cast(feature.type))
 
         self.save_dataframe(
             feature_group,
@@ -1347,17 +1343,14 @@ def save_empty_dataframe(self, feature_group, new_features=None):
             {},
         )
 
-    def add_cols_to_delta_table(self, feature_group, new_features):
+    def add_cols_to_delta_table(self, feature_group):
         location = feature_group.prepare_spark_location()
 
         dataframe = self._spark_session.read.format("delta").load(location)
 
-        if (new_features is not None):
-            if isinstance(new_features, list):
-                for new_feature in new_features:
-                    dataframe = dataframe.withColumn(new_feature.name, lit("").cast(new_feature.type))
-            else:
-                dataframe = dataframe.withColumn(new_features.name, lit("").cast(new_features.type))
+        for feature in feature_group.features:
+            if feature.name not in dataframe.columns:
+                dataframe = dataframe.withColumn(feature.name, lit(None).cast(feature.type))
 
         dataframe.limit(0).write.format("delta").mode(
             "append"