fix reconcile and prepare_spark

Author: bubriks

python/hsfs/core/feature_group_engine.py

@@ -285,7 +285,7 @@ def append_features(self, feature_group, new_features):
         if feature_group.time_travel_format == "DELTA":
             engine.get_instance().add_cols_to_delta_table(feature_group, new_features)
         else:
-            engine.get_instance().save_empty_dataframe(feature_group)
+            engine.get_instance().save_empty_dataframe(feature_group, new_features=new_features)
 
     def update_description(self, feature_group, description):
         """Updates the description of a feature group."""

python/hsfs/core/hudi_engine.py

@@ -20,6 +20,10 @@
 
 
 class HudiEngine:
+
+    HUDI_SPEC_FEATURE_NAMES = ["_hoodie_record_key", "_hoodie_partition_path",
+                               "_hoodie_commit_time", "_hoodie_file_name", "_hoodie_commit_seqno"]
+
     HUDI_SPARK_FORMAT = "org.apache.hudi"
     HUDI_TABLE_NAME = "hoodie.table.name"
     HUDI_TABLE_STORAGE_TYPE = "hoodie.datasource.write.storage.type"
@@ -229,9 +233,8 @@ def _setup_hudi_read_opts(self, hudi_fg_alias, read_options):
     def reconcile_hudi_schema(
         self, save_empty_dataframe_callback, hudi_fg_alias, read_options
     ):
-        fg_table_name = hudi_fg_alias.feature_group._get_table_name()
         if sorted(self._spark_session.table(hudi_fg_alias.alias).columns) != sorted(
-            self._spark_session.table(fg_table_name).columns
+            [feature.name for feature in hudi_fg_alias.feature_group._features] + self.HUDI_SPEC_FEATURE_NAMES
         ):
             full_fg = self._feature_group_api.get(
                 feature_store_id=hudi_fg_alias.feature_group._feature_store_id,

python/hsfs/engine/python.py

@@ -1182,7 +1182,7 @@ def save_stream_dataframe(
         )
 
     def save_empty_dataframe(
-        self, feature_group: Union[FeatureGroup, ExternalFeatureGroup]
+        self, feature_group: Union[FeatureGroup, ExternalFeatureGroup], new_features=None
     ) -> None:
         """Wrapper around save_dataframe in order to provide no-op."""
         pass

python/hsfs/engine/spark.py

@@ -219,10 +219,9 @@ def register_hudi_temporary_table(
             read_options,
         )
 
-        if (hudi_fg_alias._feature_group.storage_connector is None):
-            hudi_engine_instance.reconcile_hudi_schema(
-                self.save_empty_dataframe, hudi_fg_alias, read_options
-            )
+        hudi_engine_instance.reconcile_hudi_schema(
+            self.save_empty_dataframe, hudi_fg_alias, read_options
+        )
 
     def register_delta_temporary_table(
         self, delta_fg_alias, feature_store_id, feature_store_name, read_options
@@ -1250,13 +1249,22 @@ def is_spark_dataframe(self, dataframe):
             return True
         return False
 
-    def save_empty_dataframe(self, feature_group):
-        fg_table_name = feature_group._get_table_name()
-        dataframe = self._spark_session.table(fg_table_name).limit(0)
+    def save_empty_dataframe(self, feature_group, new_features=None):
+        dataframe = self._spark_session.read.format("hudi").load(
+            feature_group.get_uri()
+        )
+
+        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))
+
 
         self.save_dataframe(
             feature_group,
-            dataframe,
+            dataframe.limit(0),
             "upsert",
             feature_group.online_enabled,
             "offline",

python/hsfs/storage_connector.py

@@ -134,6 +134,12 @@ def description(self) -> Optional[str]:
     def spark_options(self) -> None:
         pass
 
+    def prepare_spark(self, path: Optional[str] = None) -> Optional[str]:
+        _logger.info(
+            "This Storage Connector cannot be prepare for Spark."
+        )
+        return path
+
     def read(
         self,
         query: Optional[str] = None,

python/tests/engine/test_spark.py

@@ -4559,7 +4559,9 @@ def test_save_empty_dataframe(self, mocker):
         mock_spark_engine_save_dataframe = mocker.patch(
             "hsfs.engine.spark.Engine.save_dataframe"
         )
-        mock_spark_table = mocker.patch("pyspark.sql.session.SparkSession.table")
+        mock_spark_read = mocker.patch("pyspark.sql.SparkSession.read")
+        mock_format = mocker.Mock()
+        mock_spark_read.format.return_value = mock_format
 
         # Arrange
         spark_engine = spark.Engine()
@@ -4579,7 +4581,7 @@ def test_save_empty_dataframe(self, mocker):
 
         # Assert
         assert mock_spark_engine_save_dataframe.call_count == 1
-        assert mock_spark_table.call_count == 1
+        assert mock_spark_read.format.call_count == 1
 
     def test_apply_transformation_function_single_output(self, mocker):
         # Arrange