Fix File path handling across storage backends (#1604)

* File path handling cleanup

* fix tests

* escape listing dataset names properly

* fix tests

* fix(local): make is_path_in cross-platform and add drive-letter check

Replace Path.resolve(strict=False) with os.path.abspath + os.path.normcase
in is_path_in() for deterministic, filesystem-independent path comparison.
resolve() varies across Python versions (3.11 uses abspath, 3.12+ uses
realpath) and can produce different results depending on which path
components already exist on disk.

Also use startswith(output + os.sep) instead of is_relative_to() to
prevent false positives where /foo/bar2 would match /foo/bar.

Add drive-letter absolute path check to validate_local_relpath() so
Windows paths like C:/secret/file are rejected.

* review and refactor export, fix tests again

* normalize only local windows paths

* normalize only local windows paths

* revert unrelated changes

* review, cleanup

* cleanup / fix listings handling

* remove get_full_path from the client

* remove rel path

* simplify, unify validation

* more cleanup, refactor to address review

* address PR review

* prefetch skip and log bad files

* add more test for func on path

* review, cleanup, add tests for save / upload

* more review fixes

* address one more review

Author: shcheklein

src/datachain/cache.py

@@ -78,16 +78,22 @@ async def download(
     ) -> None:
         from dvc_objects.fs.utils import tmp_fname
 
-        from_path = file.get_uri()
+        from_path = file.get_fs_path()
         odb_fs = self.odb.fs
         tmp_info = odb_fs.join(self.odb.tmp_dir, tmp_fname())  # type: ignore[arg-type]
         size = file.size
         if size < 0:
-            size = await client.get_size(from_path, version_id=file.version)
+            size = await client.get_size(file)
         from datachain.progress import tqdm
 
         cb = callback or TqdmCallback(
-            tqdm_kwargs={"desc": odb_fs.name(from_path), "bytes": True, "leave": False},
+            tqdm_kwargs={
+                "desc": odb_fs.name(from_path),
+                "unit": "B",
+                "unit_scale": True,
+                "unit_divisor": 1024,
+                "leave": False,
+            },
             tqdm_cls=tqdm,
             size=size,
         )

src/datachain/catalog/catalog.py

@@ -479,7 +479,7 @@ def find_column_to_str(  # noqa: PLR0911
             full_path = path + "/"
         else:
             full_path = path
-        return src.get_node_full_path_from_path(full_path)
+        return src.get_node_uri_from_path(full_path)
     if column == "size":
         return str(row[field_lookup["size"]])
     if column == "type":
@@ -2114,7 +2114,7 @@ def du_dirs(src, node, subdepth):
                     for sd in subdirs:
                         yield from du_dirs(src, sd, subdepth - 1)
                 yield (
-                    src.get_node_full_path(node),
+                    src.get_node_uri(node),
                     src.listing.du(node)[0],
                 )
 

src/datachain/catalog/datasource.py

@@ -12,11 +12,11 @@ def __init__(self, listing, client, node, as_container=False):
             as_container  # Indicates whether a .tar file is handled as a container
         )
 
-    def get_node_full_path(self, node):
-        return self.client.get_full_path(node.full_path)
+    def get_node_uri(self, node):
+        return self.client.get_uri(node.full_path)
 
-    def get_node_full_path_from_path(self, full_path):
-        return self.client.get_full_path(full_path)
+    def get_node_uri_from_path(self, full_path):
+        return self.client.get_uri(full_path)
 
     def is_single_object(self):
         return self.node.dir_type == DirType.FILE or (

src/datachain/client/azure.py

@@ -1,5 +1,4 @@
 from typing import Any
-from urllib.parse import parse_qs, urlsplit, urlunsplit
 
 from adlfs import AzureBlobFileSystem
 
@@ -26,19 +25,48 @@ def info_to_file(self, v: dict[str, Any], path: str) -> File:
             size=v.get("size", ""),
         )
 
-    def url(self, path: str, expires: int = 3600, **kwargs) -> str:
+    def url(
+        self,
+        path: str,
+        expires: int = 3600,
+        version_id: str | None = None,
+        **kwargs,
+    ) -> str:
         """
         Generate a signed URL for the given path.
         """
-        version_id = kwargs.pop("version_id", None)
         content_disposition = kwargs.pop("content_disposition", None)
+        full_path = self.get_uri(path)
+        if version_id:
+            # adlfs.split_path() reads version_id from the urlpath.
+            full_path = f"{full_path}?versionid={version_id}"
+
         result = self.fs.sign(
-            self.get_full_path(path, version_id),
+            full_path,
             expiration=expires,
             content_disposition=content_disposition,
             **kwargs,
         )
-        return result + (f"&versionid={version_id}" if version_id else "")
+
+        if version_id:
+            # The Azure SDK does not embed versionid in the SAS token, so we
+            # append it explicitly to route the request to the correct version.
+            result += f"&versionid={version_id}"
+        return result
+
+    async def get_file(
+        self,
+        lpath: str,
+        rpath: str,
+        callback,
+        version_id: str | None = None,
+    ) -> None:
+        if version_id:
+            # adlfs._get_file() only reads version_id from split_path(rpath);
+            # it does not accept version_id as a kwarg.  Embed it in the path
+            # so split_path can recover it on the adlfs side.
+            lpath = f"{lpath}?versionid={version_id}"
+        await self.fs._get_file(lpath, rpath, callback=callback)
 
     async def _fetch_flat(self, start_prefix: str, result_queue: ResultQueue) -> None:
         prefix = start_prefix
@@ -72,13 +100,4 @@ async def _fetch_flat(self, start_prefix: str, result_queue: ResultQueue) -> Non
         finally:
             result_queue.put_nowait(None)
 
-    @classmethod
-    def version_path(cls, path: str, version_id: str | None) -> str:
-        parts = list(urlsplit(path))
-        query = parse_qs(parts[3])
-        if "versionid" in query:
-            raise ValueError("path already includes a version query")
-        parts[3] = f"versionid={version_id}" if version_id else ""
-        return urlunsplit(parts)
-
     _fetch_default = _fetch_flat

src/datachain/client/fsspec.py

@@ -5,7 +5,6 @@
 import os
 import posixpath
 import re
-import sys
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterator, Iterator, Sequence
 from datetime import datetime
@@ -19,6 +18,7 @@
 
 from datachain.cache import Cache
 from datachain.client.fileslice import FileWrapper
+from datachain.fs.utils import is_win_local_path
 from datachain.nodes_fetcher import NodesFetcher
 from datachain.nodes_thread_pool import NodeChunk
 from datachain.progress import tqdm
@@ -41,20 +41,6 @@
 ResultQueue = asyncio.Queue[Sequence["File"] | None]
 
 
-def is_win_local_path(uri: str) -> bool:
-    if sys.platform == "win32":
-        if len(uri) >= 1 and uri[0] == "\\":
-            return True
-        if (
-            len(uri) >= 3
-            and uri[1] == ":"
-            and (uri[2] == "/" or uri[2] == "\\")
-            and uri[0].isalpha()
-        ):
-            return True
-    return False
-
-
 def is_cloud_uri(uri: str) -> bool:
     protocol = urlparse(uri).scheme
     return protocol in CLOUD_STORAGE_PROTOCOLS
@@ -82,7 +68,7 @@ def __init__(self, name: str, fs_kwargs: dict[str, Any], cache: Cache) -> None:
         self.fs_kwargs = fs_kwargs
         self._fs: AbstractFileSystem | None = None
         self.cache = cache
-        self.uri = self.get_uri(self.name)
+        self.uri = self.storage_uri(self.name)
 
     @staticmethod
     def get_implementation(url: str | os.PathLike[str]) -> type["Client"]:  # noqa: PLR0911
@@ -114,29 +100,44 @@ def get_implementation(url: str | os.PathLike[str]) -> type["Client"]:  # noqa:
 
         raise NotImplementedError(f"Unsupported protocol: {protocol}")
 
-    @classmethod
-    def path_to_uri(cls, path: str) -> str:
-        """Convert a path-like object to a URI. Default: identity."""
-        return path
-
     @staticmethod
     def is_data_source_uri(name: str) -> bool:
         # Returns True if name is one of supported data sources URIs, e.g s3 bucket
         return DATA_SOURCE_URI_PATTERN.match(name) is not None
 
+    @staticmethod
+    def validate_file_path(path: str) -> None:
+        """Validate a relative object path for this backend.
+
+        Raises ``ValueError`` for paths that are empty, end with ``/``, or
+        contain ``.`` / ``..`` segments.  Subclasses extend this with
+        backend-specific rules (e.g. local-filesystem restrictions).
+        """
+        if not path:
+            raise ValueError("path must not be empty")
+        if path.endswith("/"):
+            raise ValueError("path must not be a directory")
+        parts = path.split("/")
+        if any(part in (".", "..") for part in parts):
+            raise ValueError("path must not contain '.' or '..'")
+
+    @classmethod  # noqa: B027
+    def validate_source(cls, source: str) -> None:
+        """Validate the source URI for this backend."""
+
     @staticmethod
     def parse_url(source: str) -> tuple["StorageURI", str]:
         cls = Client.get_implementation(source)
         storage_name, rel_path = cls.split_url(source)
-        return cls.get_uri(storage_name), rel_path
+        return cls.storage_uri(storage_name), rel_path
 
     @staticmethod
     def get_client(source: str | os.PathLike[str], cache: Cache, **kwargs) -> "Client":
         cls = Client.get_implementation(source)
         storage_url, _ = cls.split_url(os.fspath(source))
         if os.name == "nt":
             storage_url = storage_url.removeprefix("/")
-
+        cls.validate_source(os.fspath(source))
         return cls.from_name(storage_url, cache, kwargs)
 
     @classmethod
@@ -146,10 +147,6 @@ def create_fs(cls, **kwargs) -> "AbstractFileSystem":
         fs.invalidate_cache()
         return fs
 
-    @classmethod
-    def version_path(cls, path: str, version_id: str | None) -> str:
-        return path
-
     @classmethod
     def from_name(
         cls,
@@ -185,10 +182,11 @@ def is_root_url(cls, url) -> bool:
         return url == cls.PREFIX
 
     @classmethod
-    def get_uri(cls, name: str) -> "StorageURI":
+    def storage_uri(cls, storage_name: str) -> "StorageURI":
+        """Build a :class:`StorageURI` by prepending the protocol to *storage_name*."""
         from datachain.dataset import StorageURI
 
-        return StorageURI(f"{cls.PREFIX}{name}")
+        return StorageURI(f"{cls.PREFIX}{storage_name}")
 
     @classmethod
     def split_url(cls, url: str) -> tuple[str, str]:
@@ -210,38 +208,47 @@ def fs(self) -> "AbstractFileSystem":
             self._fs = self.create_fs(**self.fs_kwargs)
         return self._fs
 
-    def url(self, path: str, expires: int = 3600, **kwargs) -> str:
-        return self.fs.sign(
-            self.get_full_path(path, kwargs.pop("version_id", None)),
-            expiration=expires,
-            **kwargs,
-        )
+    def url(
+        self,
+        path: str,
+        expires: int = 3600,
+        version_id: str | None = None,
+        **kwargs,
+    ) -> str:
+        self.validate_file_path(path)
+        kwargs.update(self._version_kwargs(version_id))
+        return self.fs.sign(self.get_uri(path), expiration=expires, **kwargs)
 
     async def get_current_etag(self, file: "File") -> str:
-        file_path = file.get_path_normalized()
-        kwargs = {}
-        if self._is_version_aware():
-            kwargs["version_id"] = file.version
-        info = await self.fs._info(
-            self.get_full_path(file_path, file.version), **kwargs
-        )
-        return self.info_to_file(info, file_path).etag
+        full_path = file.get_fs_path()
+        info = await self.fs._info(full_path, **self._version_kwargs(file.version))
+        return self.info_to_file(info, file.path).etag
 
     def get_file_info(self, path: str, version_id: str | None = None) -> "File":
-        info = self.fs.info(self.get_full_path(path, version_id), version_id=version_id)
+        self.validate_file_path(path)
+        full_path = self.get_uri(path)
+        info = sync(
+            get_loop(),
+            self.fs._info,
+            full_path,
+            **self._version_kwargs(version_id),
+        )
         return self.info_to_file(info, path)
 
-    async def get_size(self, path: str, version_id: str | None = None) -> int:
-        return await self.fs._size(
-            self.version_path(path, version_id), version_id=version_id
-        )
+    async def get_size(self, file: "File") -> int:
+        full_path = file.get_fs_path()
+        info = await self.fs._info(full_path, **self._version_kwargs(file.version))
+        size = info.get("size")
+        if size is None:
+            raise FileNotFoundError(full_path)
+        return int(size)
 
     async def get_file(self, lpath, rpath, callback, version_id: str | None = None):
         return await self.fs._get_file(
-            self.version_path(lpath, version_id),
+            lpath,
             rpath,
             callback=callback,
-            version_id=version_id,
+            **self._version_kwargs(version_id),
         )
 
     async def scandir(
@@ -341,6 +348,11 @@ def _is_valid_key(key: str) -> bool:
     def _is_version_aware(self) -> bool:
         return getattr(self.fs, "version_aware", False)
 
+    def _version_kwargs(self, version_id: str | None) -> dict[str, Any]:
+        if version_id:
+            return {"version_id": version_id}
+        return {}
+
     async def ls_dir(self, path):
         kwargs = {}
         if self._is_version_aware():
@@ -350,8 +362,9 @@ async def ls_dir(self, path):
     def rel_path(self, path: str) -> str:
         return self.fs.split_path(path)[1]
 
-    def get_full_path(self, rel_path: str, version_id: str | None = None) -> str:
-        return self.version_path(f"{self.PREFIX}{self.name}/{rel_path}", version_id)
+    def get_uri(self, rel_path: str) -> str:
+        """Build a full URI for the given relative path within this client's storage."""
+        return f"{self.PREFIX}{self.name}/{rel_path}"
 
     @abstractmethod
     def info_to_file(self, v: dict[str, Any], path: str) -> "File": ...
@@ -397,20 +410,29 @@ def open_object(
         if use_cache and (cache_path := self.cache.get_path(file)):
             return open(cache_path, mode="rb")
         assert not file.location
+        kwargs = self._version_kwargs(file.version)
+        full_path = file.get_fs_path()
         return FileWrapper(
-            self.fs.open(self.get_full_path(file.get_path_normalized(), file.version)),
+            self.fs.open(full_path, **kwargs),
             cb,
         )  # type: ignore[return-value]
 
     def upload(self, data: bytes, path: str) -> "File":
-        full_path = path if path.startswith(self.PREFIX) else self.get_full_path(path)
+        if path.startswith(self.PREFIX):
+            full_path = path
+            _, rel_path = self.split_url(path)
+        else:
+            rel_path = path
+            full_path = self.get_uri(path)
+
+        self.validate_file_path(rel_path)
 
         parent = posixpath.dirname(full_path)
         self.fs.makedirs(parent, exist_ok=True)
 
         self.fs.pipe_file(full_path, data)
         file_info = self.fs.info(full_path)
-        return self.info_to_file(file_info, path)
+        return self.info_to_file(file_info, rel_path)
 
     def download(self, file: "File", *, callback: Callback = DEFAULT_CALLBACK) -> None:
         sync(get_loop(), functools.partial(self._download, file, callback=callback))