fix(dask): preserve remote URIs (e.g. s3://) in DaskOfflineStore path resolution

Signed-off-by: sanghyun <[email protected]>

Author: predictorsSH

sdk/python/feast/infra/offline_stores/dask.py

@@ -521,10 +521,9 @@ def offline_write_batch(
 
         file_options = feature_view.batch_source.file_options
 
-        if config.repo_path is not None and not Path(file_options.uri).is_absolute():
-            absolute_path = config.repo_path / file_options.uri
-        else:
-            absolute_path = Path(file_options.uri)
+        absolute_path = FileSource.get_uri_for_file_path(
+            repo_path=config.repo_path, uri=file_options.uri
+        )
 
         filesystem, path = FileSource.create_filesystem_and_path(
             str(absolute_path), file_options.s3_endpoint_override
@@ -574,10 +573,11 @@ def _read_datasource(data_source, repo_path) -> dd.DataFrame:
         else None
     )
 
-    if not Path(data_source.path).is_absolute():
-        path = repo_path / data_source.path
-    else:
-        path = data_source.path
+    path = FileSource.get_uri_for_file_path(
+        repo_path=repo_path,
+        uri=data_source.file_options.uri,
+    )
+
     return dd.read_parquet(
         path,
         storage_options=storage_options,