feat(to_storage): cleanup, add tests, add filepath strategy (#1461)

Author: shcheklein

docs/guide/remotes.md

@@ -19,6 +19,29 @@ DataChain uses [fsspec](https://filesystem-spec.readthedocs.io/en/latest/) to in
 - Azure Blob Storage: `az://container-name/path/to/data`
 - Hugging Face: `hf://dataset-name`
 
+## Export placement strategies
+
+Both `File.export` and `DataChain.to_storage` expose a `placement`
+parameter that controls how the target path is built for each file. Choose the
+strategy that fits your destination layout:
+
+- `filename`: keep only the original file name (no directories). Simple, but
+    colliding names overwrite each other.
+- `filepath`: preserve the relative directory structure inside the dataset.
+    This is the right choice when you want to mirror the source folder layout at
+    the destination.
+- `fullpath`: for remote sources, prefix paths with the storage host (for
+    example `s3://bucket/data/a.jpg` exports to `<output>/bucket/data/a.jpg`).
+    Local sources behave the same as `filepath`.
+- `etag`: use the file ETag with the original extension to guarantee unique
+    names when your storage exposes object digests.
+- `checksum`: reserved for future use. Calling it currently raises
+    `NotImplementedError`.
+
+Relative output directories such as `"."` or `".."`, as well as absolute
+paths, are supported. The placement strategy simply determines the appended
+sub-path.
+
 ## Extra configuration
 For the configuration parameters to the filesystem, you can pass the key and value pair as client_config dictionary that will be passed to the respective filesystem.
 

src/datachain/lib/dc/datachain.py

@@ -2695,7 +2695,19 @@ def to_storage(
             output: Path to the target directory for exporting files.
             signal: Name of the signal to export files from.
             placement: The method to use for naming exported files.
-                The possible values are: "filename", "etag", "fullpath", and "checksum".
+                The possible values are: "filename", "etag", "fullpath",
+                "filepath", and "checksum".
+                Example path translations for an object located at
+                ``s3://bucket/data/img.jpg`` and exported to ``./out``:
+
+                - "filename" -> ``./out/img.jpg`` (no directories)
+                - "filepath" -> ``./out/data/img.jpg`` (relative path kept)
+                - "fullpath" -> ``./out/bucket/data/img.jpg`` (remote host kept)
+                - "etag" -> ``./out/<etag>.jpg`` (unique name via object digest)
+
+                Local sources behave like "filepath" for "fullpath" placement.
+                Relative destinations such as "." or ".." and absolute paths
+                are supported for every strategy.
             link_type: Method to use for exporting files.
                 Falls back to `'copy'` if symlinking fails.
             num_threads: number of threads to use for exporting files.

src/datachain/lib/file.py

@@ -42,7 +42,7 @@
 logger = logging.getLogger("datachain")
 
 # how to create file path when exporting
-ExportPlacement = Literal["filename", "etag", "fullpath", "checksum"]
+ExportPlacement = Literal["filename", "etag", "fullpath", "checksum", "filepath"]
 
 FileType = Literal["binary", "text", "image", "video", "audio"]
 EXPORT_FILES_MAX_THREADS = 5
@@ -644,6 +644,8 @@ def get_destination_path(
             source = urlparse(self.source)
             if source.scheme and source.scheme != "file":
                 path = posixpath.join(source.netloc, path)
+        elif placement == "filepath":
+            path = unquote(self.get_path_normalized())
         elif placement == "checksum":
             raise NotImplementedError("Checksum placement not implemented yet")
         else:

tests/func/test_datachain.py

@@ -7,7 +7,9 @@
 from collections.abc import Iterator
 from datetime import datetime, timedelta, timezone
 from pathlib import Path, PurePosixPath
+from typing import cast
 from unittest.mock import Mock, patch
+from urllib.parse import urlparse
 
 import numpy as np
 import pandas as pd
@@ -259,7 +261,7 @@ def test_read_file(cloud_test_catalog, use_cache):
         assert bool(file.get_local_path()) is use_cache
 
 
-@pytest.mark.parametrize("placement", ["fullpath", "filename"])
+@pytest.mark.parametrize("placement", ["fullpath", "filename", "filepath"])
 @pytest.mark.parametrize("use_map", [True, False])
 @pytest.mark.parametrize("use_cache", [True, False])
 @pytest.mark.parametrize("file_type", ["", "binary", "text"])
@@ -305,13 +307,28 @@ def test_to_storage(
         "dog4": "ruff",
     }
 
-    for file in df.to_values("file"):
+    def _expected_destination_rel(file_obj: File, placement: str) -> Path:
+        rel_path = PurePosixPath(file_obj.path).as_posix()
+
         if placement == "filename":
-            file_path = file.name
-        else:
-            file_path = file.get_full_name()
-        with open(tmp_dir / "output" / file_path) as f:
-            assert f.read() == expected[file.name]
+            return Path(file_obj.name)
+        if placement == "filepath":
+            return Path(rel_path)
+        if placement == "fullpath":
+            parsed = urlparse(file_obj.source)
+            full_rel = rel_path
+            if parsed.scheme and parsed.scheme != "file":
+                full_rel = posixpath.join(parsed.netloc, rel_path)
+            return Path(full_rel)
+        raise AssertionError(f"Unsupported placement: {placement}")
+
+    output_root = tmp_dir / "output"
+    for file_record in df.to_values("file"):
+        file_obj = cast("File", file_record)
+        destination_rel = _expected_destination_rel(file_obj, placement)
+
+        with (output_root / destination_rel).open() as f:
+            assert f.read() == expected[file_obj.name]
 
     assert mapper.call_count == len(expected)
 

tests/unit/lib/test_file.py

@@ -1,4 +1,5 @@
 import json
+import posixpath
 from pathlib import Path
 from unittest.mock import Mock
 
@@ -50,7 +51,7 @@ def test_cache_get_path(catalog: Catalog):
 def test_get_destination_path_wrong_strategy():
     file = create_file("s3://mybkt")
     with pytest.raises(ValueError):
-        file.get_destination_path("", "wrong")
+        file.get_destination_path("", "wrong")  # type: ignore[arg-type]
 
 
 def test_get_destination_path_filename_strategy():
@@ -88,6 +89,55 @@ def test_get_destination_path_fullpath_strategy_file_source(catalog, tmp_path):
     )
 
 
+def test_get_destination_path_filepath_strategy(catalog):
+    file = create_file("s3://mybkt")
+    file._set_stream(catalog, False)
+    assert (
+        file.get_destination_path("output", "filepath") == "output/dir1/dir2/test.txt"
+    )
+
+
+def test_get_destination_path_filepath_strategy_empty_output(catalog):
+    file = create_file("s3://mybkt")
+    file._set_stream(catalog, False)
+    assert file.get_destination_path("", "filepath") == "dir1/dir2/test.txt"
+
+
+@pytest.mark.parametrize("output", [".", ".."])
+@pytest.mark.parametrize(
+    ("placement", "expected"),
+    [
+        ("filename", "test.txt"),
+        ("filepath", "dir1/dir2/test.txt"),
+        ("fullpath", "mybkt/dir1/dir2/test.txt"),
+    ],
+)
+def test_get_destination_path_relative_output(catalog, output, placement, expected):
+    file = create_file("s3://mybkt")
+    file._set_stream(catalog, False)
+    assert file.get_destination_path(output, placement) == posixpath.join(
+        output, expected
+    )
+
+
+@pytest.mark.parametrize(
+    ("placement", "expected_suffix"),
+    [
+        ("filename", "test.txt"),
+        ("filepath", "dir1/dir2/test.txt"),
+        ("fullpath", "mybkt/dir1/dir2/test.txt"),
+    ],
+)
+def test_get_destination_path_absolute_output(
+    catalog, tmp_path, placement, expected_suffix
+):
+    file = create_file("s3://mybkt")
+    file._set_stream(catalog, False)
+    output = tmp_path / "dest"
+    expected = f"{output.as_posix()}/{expected_suffix}"
+    assert file.get_destination_path(output.as_posix(), placement) == expected
+
+
 def test_read_binary_data(tmp_path, catalog: Catalog):
     file_name = "myfile"
     data = b"some\x00data\x00is\x48\x65\x6c\x57\x6f\x72\x6c\x64\xff\xffheRe"
@@ -376,7 +426,13 @@ def test_export_with_symlink(tmp_path, catalog, use_cache):
     file.export(tmp_path / "dir", link_type="symlink", use_cache=use_cache)
     assert (tmp_path / "dir" / "myfile.txt").is_symlink()
 
-    dst = Path(file.get_local_path()) if use_cache else path
+    if use_cache:
+        cached_path = file.get_local_path()
+        assert cached_path is not None
+        dst = Path(cached_path)
+    else:
+        dst = path
+
     assert (tmp_path / "dir" / "myfile.txt").resolve() == dst