small fixes

* tmp1

* tmp 2

* tmp 3

* prepare_spark

Author: bubriks

python/hsfs/constructor/fs_query.py

@@ -99,21 +99,16 @@ def register_external(
             Union[TypeVar("pyspark.sql.DataFrame"), TypeVar("pyspark.RDD")]
         ] = None,
     ) -> None:
-        if self._on_demand_fg_aliases is not None:
-            for external_fg_alias in self._on_demand_fg_aliases:
-                if type(external_fg_alias.on_demand_feature_group).__name__ == "SpineGroup":
-                    external_fg_alias.on_demand_feature_group.dataframe = spine
-                engine.get_instance().register_external_temporary_table(
-                    external_fg_alias.on_demand_feature_group,
-                    external_fg_alias.alias,
-                )
-
-        if self._hudi_cached_feature_groups is not None:
-            for external_fg_alias in self._hudi_cached_feature_groups:
-                engine.get_instance().register_external_temporary_table(
-                    external_fg_alias.feature_group,
-                    external_fg_alias.alias,
-                )
+        if self._on_demand_fg_aliases is None:
+            return
+
+        for external_fg_alias in self._on_demand_fg_aliases:
+            if type(external_fg_alias.on_demand_feature_group).__name__ == "SpineGroup":
+                external_fg_alias.on_demand_feature_group.dataframe = spine
+            engine.get_instance().register_external_temporary_table(
+                external_fg_alias.on_demand_feature_group,
+                external_fg_alias.alias,
+            )
 
     def register_hudi_tables(
         self,

python/hsfs/engine/spark.py

@@ -188,22 +188,15 @@ def set_job_group(self, group_id, description):
         self._spark_session.sparkContext.setJobGroup(group_id, description)
 
     def register_external_temporary_table(self, external_fg, alias):
-        if isinstance(external_fg, fg_mod.ExternalFeatureGroup):
+        if not isinstance(external_fg, fg_mod.SpineGroup):
             external_dataset = external_fg.storage_connector.read(
                 external_fg.query,
                 external_fg.data_format,
                 external_fg.options,
                 external_fg.get_uri(),
             )
-        elif isinstance(external_fg, fg_mod.SpineGroup):
-            external_dataset = external_fg.dataframe
         else:
-            external_dataset = external_fg.storage_connector.read(
-                None,
-                external_fg.time_travel_format,
-                None,
-                external_fg.get_uri(),
-            )
+            external_dataset = external_fg.dataframe
         if external_fg.location:
             self._spark_session.sparkContext.textFile(external_fg.location).collect()
 

python/hsfs/feature_group.py

@@ -2074,7 +2074,7 @@ def get_uri(self) -> str:
         else:
             path = self.storage_connector._get_path(self.path)
             if engine.get_type().startswith("spark"):
-                path = engine.get_instance().setup_storage_connector(self.storage_connector, path)
+                path = self.storage_connector.prepare_spark(path)
             return path
 
     @property