fix(signed url): pass version id down to fsspec via path

src/datachain/catalog/catalog.py

@@ -1236,10 +1236,16 @@ def ls_dataset_rows(
 
         return q.to_db_records()
 
-    def signed_url(self, source: str, path: str, client_config=None) -> str:
+    def signed_url(
+        self,
+        source: str,
+        path: str,
+        version_id: Optional[str] = None,
+        client_config=None,
+    ) -> str:
         client_config = client_config or self.client_config
         client = Client.get_client(source, self.cache, **client_config)
-        return client.url(path)
+        return client.url(path, version_id=version_id)
 
     def export_dataset_table(
         self,

src/datachain/client/fsspec.py

@@ -202,7 +202,11 @@ def fs(self) -> "AbstractFileSystem":
         return self._fs
 
     def url(self, path: str, expires: int = 3600, **kwargs) -> str:
-        return self.fs.sign(self.get_full_path(path), expiration=expires, **kwargs)
+        return self.fs.sign(
+            self.get_full_path(path, kwargs.pop("version_id", None)),
+            expiration=expires,
+            **kwargs,
+        )
 
     async def get_current_etag(self, file: "File") -> str:
         kwargs = {}

src/datachain/client/s3.py

@@ -1,5 +1,6 @@
 import asyncio
 from typing import Any, Optional, cast
+from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 from botocore.exceptions import NoCredentialsError
 from s3fs import S3FileSystem
@@ -121,6 +122,32 @@ def _entry_from_boto(self, v, bucket, versions=False) -> File:
             size=v["Size"],
         )
 
+    @classmethod
+    def _split_version(cls, path: str) -> tuple[str, Optional[str]]:
+        parts = list(urlsplit(path))
+        query = parse_qs(parts[3])
+        if "versionId" in query:
+            version_id = query["versionId"][0]
+            del query["versionId"]
+            parts[3] = urlencode(query)
+        else:
+            version_id = None
+        return urlunsplit(parts), version_id
+
+    @classmethod
+    def _join_version(cls, path: str, version_id: Optional[str]) -> 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)
+
+    @classmethod
+    def version_path(cls, path: str, version_id: Optional[str]) -> str:
+        path, _ = cls._split_version(path)
+        return cls._join_version(path, version_id)
+
     async def _fetch_dir(
         self,
         prefix,