get_uri()

Author: bubriks

python/hsfs/core/delta_engine.py

@@ -55,7 +55,7 @@ def register_temporary_table(self, delta_fg_alias, read_options):
         delta_options = self._setup_delta_read_opts(delta_fg_alias, read_options)
         self._spark_session.read.format(self.DELTA_SPARK_FORMAT).options(
             **delta_options
-        ).load(self._feature_group.uri).createOrReplaceTempView(
+        ).load(self._feature_group.get_uri()).createOrReplaceTempView(
             delta_fg_alias.alias
         )
 
@@ -86,14 +86,14 @@ def _setup_delta_read_opts(self, delta_fg_alias, read_options):
 
     def delete_record(self, delete_df):
         if not DeltaTable.isDeltaTable(
-            self._spark_session, self._feature_group.uri
+            self._spark_session, self._feature_group.get_uri()
         ):
             raise FeatureStoreException(
                 f"This is no data available in Feature group {self._feature_group.name}, or it not DELTA enabled "
             )
         else:
             fg_source_table = DeltaTable.forPath(
-                self._spark_session, self._feature_group.uri
+                self._spark_session, self._feature_group.get_uri()
             )
 
             source_alias = (
@@ -109,7 +109,7 @@ def delete_record(self, delete_df):
             ).whenMatchedDelete().execute()
 
         fg_commit = self._get_last_commit_metadata(
-            self._spark_session, self._feature_group.uri
+            self._spark_session, self._feature_group.get_uri()
         )
         return self._feature_group_api.commit(self._feature_group, fg_commit)
 
@@ -118,7 +118,7 @@ def _write_delta_dataset(self, dataset, write_options):
             write_options = {}
 
         if not DeltaTable.isDeltaTable(
-            self._spark_session, self._feature_group.uri
+            self._spark_session, self._feature_group.get_uri()
         ):
             (
                 dataset.write.format(DeltaEngine.DELTA_SPARK_FORMAT)
@@ -129,11 +129,11 @@ def _write_delta_dataset(self, dataset, write_options):
                     else []
                 )
                 .mode("append")
-                .save(self._feature_group.uri)
+                .save(self._feature_group.get_uri())
             )
         else:
             fg_source_table = DeltaTable.forPath(
-                self._spark_session, self._feature_group.uri
+                self._spark_session, self._feature_group.get_uri()
             )
 
             source_alias = (
@@ -149,7 +149,7 @@ def _write_delta_dataset(self, dataset, write_options):
             ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
 
         return self._get_last_commit_metadata(
-            self._spark_session, self._feature_group.uri
+            self._spark_session, self._feature_group.get_uri()
         )
 
     def _generate_merge_query(self, source_alias, updates_alias):

python/hsfs/core/hudi_engine.py

@@ -103,18 +103,18 @@ def register_temporary_table(self, hudi_fg_alias, read_options):
         hudi_options = self._setup_hudi_read_opts(hudi_fg_alias, read_options)
         self._spark_session.read.format(self.HUDI_SPARK_FORMAT).options(
             **hudi_options
-        ).load(self._feature_group.uri).createOrReplaceTempView(
+        ).load(self._feature_group.get_uri()).createOrReplaceTempView(
             hudi_fg_alias.alias
         )
 
     def _write_hudi_dataset(self, dataset, save_mode, operation, write_options):
         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.uri)
+        ).save(self._feature_group.get_uri())
 
         feature_group_commit = self._get_last_commit_metadata(
-            self._spark_context, self._feature_group.uri
+            self._spark_context, self._feature_group.get_uri()
         )
 
         return feature_group_commit

python/hsfs/engine/spark.py

@@ -193,7 +193,7 @@ def register_external_temporary_table(self, external_fg, alias):
                 external_fg.query,
                 external_fg.data_format,
                 external_fg.options,
-                external_fg.uri,
+                external_fg.get_uri(),
             )
         elif isinstance(external_fg, fg_mod.SpineGroup):
             external_dataset = external_fg.dataframe
@@ -202,7 +202,7 @@ def register_external_temporary_table(self, external_fg, alias):
                 None,
                 external_fg.time_travel_format,
                 None,
-                external_fg.uri,
+                external_fg.get_uri(),
             )
         if external_fg.location:
             self._spark_session.sparkContext.textFile(external_fg.location).collect()
@@ -1267,12 +1267,12 @@ def add_cols_to_delta_table(self, feature_group, new_features):
             new_features_map[new_features.name] = lit("").cast(new_features.type)
 
         self._spark_session.read.format("delta").load(
-            feature_group.uri
+            feature_group.get_uri()
         ).withColumns(new_features_map).limit(0).write.format("delta").mode(
             "append"
         ).option("mergeSchema", "true").option(
             "spark.databricks.delta.schema.autoMerge.enabled", "true"
-        ).save(feature_group.uri)
+        ).save(feature_group.get_uri())
 
     def _apply_transformation_function(
         self,

python/hsfs/feature_group.py

@@ -2064,8 +2064,7 @@ def path(self) -> Optional[str]:
     def storage_connector(self) -> "sc.StorageConnector":
         return self._storage_connector
 
-    @property
-    def uri(self) -> str:
+    def get_uri(self) -> str:
         """Location of data."""
         if (self.storage_connector is None):
             return self.location