[HWORKS-2799] Use feature group location to detect HopsFS storage for sink enabled feature groups (#958)

Author: gibchikafa

python/hsfs/feature_group.py

@@ -27,6 +27,7 @@
     Literal,
     TypeVar,
 )
+from urllib.parse import urlparse
 
 import avro.schema
 import hsfs.expectation_suite
@@ -3105,7 +3106,20 @@ def _init_time_travel_and_stream(
             )
 
     def _is_hopsfs_storage(self) -> bool:
-        """Return True if storage is HopsFS."""
+        """Return True if the offline storage location is HopsFS.
+
+        Sink-enabled feature groups can keep the source storage connector
+        (for example Redshift) attached while their offline data still lives
+        on the default HopsFS warehouse path.
+        In that case the location is the reliable signal for how delta-rs
+        should talk to storage.
+        """
+        location = getattr(self, "location", None)
+        if isinstance(location, str):
+            scheme = urlparse(location).scheme
+            if scheme in {"hopsfs", "hdfs"}:
+                return True
+
         return self.storage_connector is None or (
             self.storage_connector is not None
             and self.storage_connector.type == sc.StorageConnector.HOPSFS

python/tests/test_feature_group.py

@@ -944,6 +944,34 @@ def test_is_hopsfs_storage(self, sc, expected):
         fg.storage_connector = sc
         assert fg._is_hopsfs_storage() is expected
 
+    def test_is_hopsfs_storage_uses_location_when_source_connector_is_external(
+        self,
+    ):
+        fg = get_test_feature_group()
+        fg._location = (
+            "hopsfs://rpc.namenode.service.consul:8020/apps/hive/warehouse/fs.db/fg_1"
+        )
+        fg.storage_connector = storage_connector.RedshiftConnector(
+            id=1,
+            name="redshift",
+            featurestore_id=1,
+        )
+
+        assert fg._is_hopsfs_storage() is True
+
+    def test_is_hopsfs_storage_uses_single_slash_hopsfs_location(
+        self,
+    ):
+        fg = get_test_feature_group()
+        fg._location = "hopsfs:/apps/hive/warehouse/fs.db/fg_1"
+        fg.storage_connector = storage_connector.RedshiftConnector(
+            id=1,
+            name="redshift",
+            featurestore_id=1,
+        )
+
+        assert fg._is_hopsfs_storage() is True
+
     def test_init_time_travel_and_stream_uses_resolvers_python(
         self, mocker, monkeypatch
     ):