temp

python/hsfs/constructor/fs_query.py

@@ -99,16 +99,21 @@ def register_external(
             Union[TypeVar("pyspark.sql.DataFrame"), TypeVar("pyspark.RDD")]
         ] = None,
     ) -> None:
-        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,
-            )
+        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,
+                )
 
     def register_hudi_tables(
         self,

python/hsfs/core/hudi_engine.py

@@ -108,13 +108,18 @@ def register_temporary_table(self, hudi_fg_alias, read_options):
         )
 
     def _write_hudi_dataset(self, dataset, save_mode, operation, write_options):
+        if (self._feature_group.storage_connector is None):
+            location = self._feature_group.location
+        else:
+            location = self._feature_group.storage_connector._get_path(self._feature_group.path)
+
         hudi_options = self._setup_hudi_write_opts(operation, write_options)
         dataset.write.format(HudiEngine.HUDI_SPARK_FORMAT).options(**hudi_options).mode(
             save_mode
-        ).save(self._feature_group.location)
+        ).save(location)
 
         feature_group_commit = self._get_last_commit_metadata(
-            self._spark_context, self._feature_group.location
+            self._spark_context, location
         )
 
         return feature_group_commit

python/hsfs/engine/spark.py

@@ -188,13 +188,20 @@ 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 not isinstance(external_fg, fg_mod.SpineGroup):
+        if isinstance(external_fg, fg_mod.ExternalFeatureGroup):
             external_dataset = external_fg.storage_connector.read(
                 external_fg.query,
                 external_fg.data_format,
                 external_fg.options,
                 external_fg.storage_connector._get_path(external_fg.path),
             )
+        elif isinstance(external_fg, fg_mod.FeatureGroup):
+            external_dataset = external_fg.storage_connector.read(
+                None,
+                external_fg.timeTravelFormat,
+                None,
+                external_fg.storage_connector._get_path(external_fg.path),
+            )
         else:
             external_dataset = external_fg.dataframe
         if external_fg.location: