Repository: fsspec/filesystem_spec
Pull request #1833 (2025-04-30 06:29:01)

Improve CachingFileSystem performance
We cache the size attribute of the file in the metadata to be able to pass it to the open call in the CachingFileSystem. This avoids retrieving the size on each file access, which is costly for files read over a network, e.g. S3.

Closes https://github.com/fsspec/filesystem_spec/issues/1832.


Patch:
diff --git a/fsspec/generic.py b/fsspec/generic.py
index 9bad0f0..af3677e 100644
--- a/fsspec/generic.py
+++ b/fsspec/generic.py
@@ -16,15 +16,13 @@ logger = logging.getLogger("fsspec.generic")
 
 
 def set_generic_fs(protocol, **storage_options):
+    """Populate the dict used for method=="generic" lookups"""
     _generic_fs[protocol] = filesystem(protocol, **storage_options)
 
 
-default_method = "default"
-
-
-def _resolve_fs(url, method=None, protocol=None, storage_options=None):
+def _resolve_fs(url, method, protocol=None, storage_options=None):
     """Pick instance of backend FS"""
-    method = method or default_method
+    url = url[0] if isinstance(url, (list, tuple)) else url
     protocol = protocol or split_protocol(url)[0]
     storage_options = storage_options or {}
     if method == "default":
@@ -159,7 +157,7 @@ class GenericFileSystem(AsyncFileSystem):
 
     protocol = "generic"  # there is no real reason to ever use a protocol with this FS
 
-    def __init__(self, default_method="default", **kwargs):
+    def __init__(self, default_method="default", storage_options=None, **kwargs):
         """
 
         Parameters
@@ -171,22 +169,25 @@ class GenericFileSystem(AsyncFileSystem):
               configured via the config system
             - "generic": takes instances from the `_generic_fs` dict in this module,
               which you must populate before use. Keys are by protocol
+            - "options": expects storage_options, a dict mapping protocol to
+              kwargs to use when constructing the filesystem
             - "current": takes the most recently instantiated version of each FS
         """
         self.method = default_method
+        self.st_opts = storage_options
         super().__init__(**kwargs)
 
     def _parent(self, path):
-        fs = _resolve_fs(path, self.method)
+        fs = _resolve_fs(path, self.method, storage_options=self.st_opts)
         return fs.unstrip_protocol(fs._parent(path))
 
     def _strip_protocol(self, path):
         # normalization only
-        fs = _resolve_fs(path, self.method)
+        fs = _resolve_fs(path, self.method, storage_options=self.st_opts)
         return fs.unstrip_protocol(fs._strip_protocol(path))
 
     async def _find(self, path, maxdepth=None, withdirs=False, detail=False, **kwargs):
-        fs = _resolve_fs(path, self.method)
+        fs = _resolve_fs(path, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             out = await fs._find(
                 path, maxdepth=maxdepth, withdirs=withdirs, detail=True, **kwargs
@@ -251,7 +252,7 @@ class GenericFileSystem(AsyncFileSystem):
         value,
         **kwargs,
     ):
-        fs = _resolve_fs(path, self.method)
+        fs = _resolve_fs(path, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             return await fs._pipe_file(path, value, **kwargs)
         else:
@@ -269,7 +270,7 @@ class GenericFileSystem(AsyncFileSystem):
 
     async def _makedirs(self, path, exist_ok=False):
         logger.debug("Make dir %s", path)
-        fs = _resolve_fs(path, self.method)
+        fs = _resolve_fs(path, self.method, storage_options=self.st_opts)
         if fs.async_impl:
             await fs._makedirs(path, exist_ok=exist_ok)
         else:
@@ -288,6 +289,7 @@ class GenericFileSystem(AsyncFileSystem):
         url2,
         blocksize=2**20,
         callback=DEFAULT_CALLBACK,
+        tempdir: Optional[str] = None,
         **kwargs,
     ):
         fs = _resolve_fs(url, self.method)
@@ -295,35 +297,10 @@ class GenericFileSystem(AsyncFileSystem):
         if fs is fs2:
             # pure remote
             if fs.async_impl:
-                return await fs._cp_file(url, url2, **kwargs)
+                return await fs._copy(url, url2, **kwargs)
             else:
-                return fs.cp_file(url, url2, **kwargs)
-        kw = {"blocksize": 0, "cache_type": "none"}
-        try:
-            f1 = (
-                await fs.open_async(url, "rb")
-                if hasattr(fs, "open_async")
-                else fs.open(url, "rb", **kw)
-            )
-            callback.set_size(await maybe_await(f1.size))
-            f2 = (
-                await fs2.open_async(url2, "wb")
-                if hasattr(fs2, "open_async")
-                else fs2.open(url2, "wb", **kw)
-            )
-            while f1.size is None or f2.tell() < f1.size:
-                data = await maybe_await(f1.read(blocksize))
-                if f1.size is None and not data:
-                    break
-                await maybe_await(f2.write(data))
-                callback.absolute_update(f2.tell())
-        finally:
-            try:
-                await maybe_await(f2.close())
-                await maybe_await(f1.close())
-            except NameError:
-                # fail while opening f1 or f2
-                pass
+                return fs.copy(url, url2, **kwargs)
+        await copy_file_op(fs, [url], fs2, [url2], tempdir, 1, on_error="raise")
 
     async def _make_many_dirs(self, urls, exist_ok=True):
         fs = _resolve_fs(urls[0], self.method)
@@ -347,17 +324,22 @@ class GenericFileSystem(AsyncFileSystem):
         tempdir: Optional[str] = None,
         **kwargs,
     ):
+        # TODO: special case for one FS being local, which can use get/put
+        # TODO: special case for one being memFS, which can use cat/pipe
         if recursive:
-            raise NotImplementedError
-        fs = _resolve_fs(path1[0], self.method)
-        fs2 = _resolve_fs(path2[0], self.method)
-        # not expanding paths atm., assume call is from rsync()
+            raise NotImplementedError("Please use fsspec.generic.rsync")
+        path1 = [path1] if isinstance(path1, str) else path1
+        path2 = [path2] if isinstance(path2, str) else path2
+
+        fs = _resolve_fs(path1, self.method)
+        fs2 = _resolve_fs(path2, self.method)
+
         if fs is fs2:
-            # pure remote
             if fs.async_impl:
                 return await fs._copy(path1, path2, **kwargs)
             else:
                 return fs.copy(path1, path2, **kwargs)
+
         await copy_file_op(
             fs, path1, fs2, path2, tempdir, batch_size, on_error=on_error
         )
@@ -377,31 +359,33 @@ async def copy_file_op(
                 fs2,
                 u2,
                 os.path.join(tempdir, uuid.uuid4().hex),
-                on_error=on_error,
             )
             for u1, u2 in zip(url1, url2)
         ]
-        await _run_coros_in_chunks(coros, batch_size=batch_size)
+        out = await _run_coros_in_chunks(
+            coros, batch_size=batch_size, return_exceptions=True
+        )
     finally:
         shutil.rmtree(tempdir)
+    if on_error == "return":
+        return out
+    elif on_error == "raise":
+        for o in out:
+            if isinstance(o, Exception):
+                raise o
 
 
 async def _copy_file_op(fs1, url1, fs2, url2, local, on_error="ignore"):
-    ex = () if on_error == "raise" else Exception
-    logger.debug("Copy %s -> %s", url1, url2)
-    try:
-        if fs1.async_impl:
-            await fs1._get_file(url1, local)
-        else:
-            fs1.get_file(url1, local)
-        if fs2.async_impl:
-            await fs2._put_file(local, url2)
-        else:
-            fs2.put_file(local, url2)
-        os.unlink(local)
-        logger.debug("Copy %s -> %s; done", url1, url2)
-    except ex as e:
-        logger.debug("ignoring cp exception for %s: %s", url1, e)
+    if fs1.async_impl:
+        await fs1._get_file(url1, local)
+    else:
+        fs1.get_file(url1, local)
+    if fs2.async_impl:
+        await fs2._put_file(local, url2)
+    else:
+        fs2.put_file(local, url2)
+    os.unlink(local)
+    logger.debug("Copy %s -> %s; done", url1, url2)
 
 
 async def maybe_await(cor):
diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index bd58ad4..929ec1f 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -335,6 +335,11 @@ class CachingFileSystem(AbstractFileSystem):
             self._metadata.update_file(path, detail)
             logger.debug("Creating local sparse file for %s", path)
 
+        # explicitly submitting the size to the open call will avoid extra
+        # operations when opening. This is particularly relevant
+        # for any file that is read over a network, e.g. S3.
+        size = detail.get("size", None)
+
         # call target filesystems open
         self._mkcache()
         f = self.fs._open(
@@ -344,8 +349,15 @@ class CachingFileSystem(AbstractFileSystem):
             autocommit=autocommit,
             cache_options=cache_options,
             cache_type="none",
+            size=size,
             **kwargs,
         )
+
+        # set size if not already set
+        if size is None:
+            detail["size"] = f.size
+            self._metadata.update_file(path, detail)
+
         if self.compression:
             comp = (
                 infer_compression(path)
diff --git a/fsspec/implementations/dbfs.py b/fsspec/implementations/dbfs.py
index fa71e1a..2c56a42 100644
--- a/fsspec/implementations/dbfs.py
+++ b/fsspec/implementations/dbfs.py
@@ -14,12 +14,13 @@ class DatabricksException(Exception):
     Helper class for exceptions raised in this module.
     """
 
-    def __init__(self, error_code, message):
+    def __init__(self, error_code, message, details=None):
         """Create a new DatabricksException"""
         super().__init__(message)
 
         self.error_code = error_code
         self.message = message
+        self.details = details
 
 
 class DatabricksFileSystem(AbstractFileSystem):
@@ -80,7 +81,7 @@ class DatabricksFileSystem(AbstractFileSystem):
                     raise FileNotFoundError(e.message) from e
 
                 raise
-            files = r["files"]
+            files = r.get("files", [])
             out = [
                 {
                     "name": o["path"],
diff --git a/fsspec/implementations/local.py b/fsspec/implementations/local.py
index 70dfd90..0152816 100644
--- a/fsspec/implementations/local.py
+++ b/fsspec/implementations/local.py
@@ -430,12 +430,22 @@ class LocalFileOpener(io.IOBase):
     def commit(self):
         if self.autocommit:
             raise RuntimeError("Can only commit if not already set to autocommit")
-        shutil.move(self.temp, self.path)
         try:
-            mask = 0o666
-            os.chmod(self.path, mask & ~get_umask(mask))
-        except RuntimeError:
-            pass
+            shutil.move(self.temp, self.path)
+        except PermissionError as e:
+            # shutil.move raises PermissionError if os.rename
+            # and the default copy2 fallback with shutil.copystats fail.
+            # The file should be there nonetheless, but without copied permissions.
+            # If it doesn't exist, there was no permission to create the file.
+            if not os.path.exists(self.path):
+                raise e
+        else:
+            # If PermissionError is not raised, permissions can be set.
+            try:
+                mask = 0o666
+                os.chmod(self.path, mask & ~get_umask(mask))
+            except RuntimeError:
+                pass
 
     def discard(self):
         if self.autocommit:
diff --git a/pyproject.toml b/pyproject.toml
index abe106c..c00e306 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -183,7 +183,7 @@ select = [
     "SIM",
     "SLOT",
     "SIM101",
-    "TCH",
+    "TC",
     "UP",
 ]
 ignore = [


Test patch:
diff --git a/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_file_listing.yaml b/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_file_listing.yaml
deleted file mode 100644
index 57929bf..0000000
--- a/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_file_listing.yaml
+++ /dev/null
@@ -1,112 +0,0 @@
-interactions:
-- request:
-    body: '{"path": "/"}'
-    headers:
-      Accept:
-      - '*/*'
-      Accept-Encoding:
-      - gzip, deflate
-      Connection:
-      - keep-alive
-      Content-Length:
-      - '13'
-      Content-Type:
-      - application/json
-      User-Agent:
-      - python-requests/2.31.0
-      authorization:
-      - DUMMY
-    method: GET
-    uri: https://my_instance.com/api/2.0/dbfs/list
-  response:
-    body:
-      string: !!binary |
-        H4sIAAAAAAAEA6zSSwrCMBAG4LvMumLSpo3JAbyA4Eak1DbiYF8kExeW3t2qC6G46WM3/AwfzKOD
-        K5bGgT510GZ0Aw3b/ZAcqLEGAkCXFmhBk/Um+PSmDp8GNAugagq8Yp4RNnVKWA0pl0woxmMhGGN9
-        8DOPTemHhsngH8QtVIqMsovF/O4279IZWlG0xvlyMVjVNH1InoTD5kMpotHuqWrnaErGkeJyN9K8
-        M3YOxxmLEhGqEfdY4zG+yKwznvsXAAAA//8DAMDPrToDAwAA
-    headers:
-      access-control-allow-headers:
-      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
-        Content-Type
-      access-control-allow-origin:
-      - '*'
-      cache-control:
-      - no-cache, no-store, must-revalidate
-      content-encoding:
-      - gzip
-      content-type:
-      - application/json
-      expires:
-      - '0'
-      pragma:
-      - no-cache
-      server:
-      - databricks
-      strict-transport-security:
-      - max-age=31536000; includeSubDomains; preload
-      transfer-encoding:
-      - chunked
-      vary:
-      - Accept-Encoding
-      x-content-type-options:
-      - nosniff
-    status:
-      code: 200
-      message: OK
-- request:
-    body: '{"path": "/"}'
-    headers:
-      Accept:
-      - '*/*'
-      Accept-Encoding:
-      - gzip, deflate
-      Connection:
-      - keep-alive
-      Content-Length:
-      - '13'
-      Content-Type:
-      - application/json
-      User-Agent:
-      - python-requests/2.31.0
-      authorization:
-      - DUMMY
-    method: GET
-    uri: https://my_instance.com/api/2.0/dbfs/list
-  response:
-    body:
-      string: !!binary |
-        H4sIAAAAAAAEA6zSSwrCMBAG4LvMumLSpo3JAbyA4Eak1DbiYF8kExeW3t2qC6G46WM3/AwfzKOD
-        K5bGgT510GZ0Aw3b/ZAcqLEGAkCXFmhBk/Um+PSmDp8GNAugagq8Yp4RNnVKWA0pl0woxmMhGGN9
-        8DOPTemHhsngH8QtVIqMsovF/O4279IZWlG0xvlyMVjVNH1InoTD5kMpotHuqWrnaErGkeJyN9K8
-        M3YOxxmLEhGqEfdY4zG+yKwznvsXAAAA//8DAMDPrToDAwAA
-    headers:
-      access-control-allow-headers:
-      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
-        Content-Type
-      access-control-allow-origin:
-      - '*'
-      cache-control:
-      - no-cache, no-store, must-revalidate
-      content-encoding:
-      - gzip
-      content-type:
-      - application/json
-      expires:
-      - '0'
-      pragma:
-      - no-cache
-      server:
-      - databricks
-      strict-transport-security:
-      - max-age=31536000; includeSubDomains; preload
-      transfer-encoding:
-      - chunked
-      vary:
-      - Accept-Encoding
-      x-content-type-options:
-      - nosniff
-    status:
-      code: 200
-      message: OK
-version: 1
diff --git a/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_listing.yaml b/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_listing.yaml
new file mode 100644
index 0000000..866f32a
--- /dev/null
+++ b/fsspec/implementations/tests/cassettes/test_dbfs/test_dbfs_listing.yaml
@@ -0,0 +1,214 @@
+interactions:
+- request:
+    body: '{"path": "/FileStore/fsspec_dbfs/project_root/empty_dir"}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '57'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.32.3
+      authorization:
+      - DUMMY
+    method: POST
+    uri: https://my_instance.com/api/2.0/dbfs/mkdirs
+  response:
+    body:
+      string: '{}'
+    headers:
+      access-control-allow-headers:
+      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
+        Content-Type
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - no-cache, no-store, must-revalidate
+      content-length:
+      - '2'
+      content-type:
+      - application/json
+      expires:
+      - '0'
+      pragma:
+      - no-cache
+      server:
+      - databricks
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      x-content-type-options:
+      - nosniff
+      x-request-id:
+      - e71bb75f-b05a-4523-b8bc-ce331812fbf0
+    status:
+      code: 200
+      message: OK
+- request:
+    body: '{"path": "/FileStore/fsspec_dbfs"}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '34'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.32.3
+      authorization:
+      - DUMMY
+    method: GET
+    uri: https://my_instance.com/api/2.0/dbfs/list
+  response:
+    body:
+      string: !!binary |
+        H4sIAAAAAAAEAxzMMQ7CMAwF0Lv8OVKYcwAuwIiQ1SaOMKI4ss3SqncvZX3D29DlzY5y3zCmeKIg
+        X39yCzXO3X1wpTZ3z8P0xTXIVAMJ4tTEUMK+nP4JuayMcklYtEmXOoXoh0KWU/fHfgAAAP//AwCB
+        uORhbAAAAA==
+    headers:
+      access-control-allow-headers:
+      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
+        Content-Type
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - no-cache, no-store, must-revalidate
+      content-encoding:
+      - gzip
+      content-type:
+      - application/json
+      expires:
+      - '0'
+      pragma:
+      - no-cache
+      server:
+      - databricks
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      transfer-encoding:
+      - chunked
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+      x-request-id:
+      - 6e66514c-c883-49c7-b215-2fecc5ef8222
+    status:
+      code: 200
+      message: OK
+- request:
+    body: '{"path": "/FileStore/fsspec_dbfs/project_root/empty_dir"}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '57'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.32.3
+      authorization:
+      - DUMMY
+    method: GET
+    uri: https://my_instance.com/api/2.0/dbfs/list
+  response:
+    body:
+      string: '{}'
+    headers:
+      access-control-allow-headers:
+      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
+        Content-Type
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - no-cache, no-store, must-revalidate
+      content-length:
+      - '2'
+      content-type:
+      - application/json
+      expires:
+      - '0'
+      pragma:
+      - no-cache
+      server:
+      - databricks
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      x-content-type-options:
+      - nosniff
+      x-request-id:
+      - 59b8c763-be19-4402-8d37-135a7d6c7aed
+    status:
+      code: 200
+      message: OK
+- request:
+    body: '{"path": "/FileStore/fsspec_dbfs/project_root/missing_directory"}'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '65'
+      Content-Type:
+      - application/json
+      User-Agent:
+      - python-requests/2.32.3
+      authorization:
+      - DUMMY
+    method: GET
+    uri: https://my_instance.com/api/2.0/dbfs/list
+  response:
+    body:
+      string: !!binary |
+        H4sIAAAAAAAEAzyOwWrDMBBEf0XoXNtx4hqSU6B1IZcY4hQKpQhFWrkqtlfdVUtDyL9XbqCnZdh5
+        M3ORQISkDFqQG3louvb58NCox7bp1L49quZl1x3lnRyBWfezZ4/C+QEEkrCewESks4Afz5EFTiLo
+        +C6Kp+To0gcKxxzAKHtyXATCjwQoQozF6Jn91Kv/kDzVWIjaDyw3rxe5jecwF84n7xH7AXTwnBsc
+        i5vMKZj8AJ9fwHE3OUwBdFPK20TWTjuoa8jK2q6yamGq7ATrZbYuYelW9aLU91ViGOj7b4mOOlHy
+        +nb9BQAA//8DADr+FHAYAQAA
+    headers:
+      access-control-allow-headers:
+      - Authorization, X-Databricks-Azure-Workspace-Resource-Id, X-Databricks-Org-Id,
+        Content-Type
+      access-control-allow-origin:
+      - '*'
+      cache-control:
+      - no-cache, no-store, must-revalidate
+      content-encoding:
+      - gzip
+      content-type:
+      - application/json
+      expires:
+      - '0'
+      pragma:
+      - no-cache
+      server:
+      - databricks
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      transfer-encoding:
+      - chunked
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+      x-request-id:
+      - 6fafe66e-16d3-40c4-be92-91e2f3601a54
+    status:
+      code: 404
+      message: Not Found
+version: 1
diff --git a/fsspec/implementations/tests/test_dbfs.py b/fsspec/implementations/tests/test_dbfs.py
index 66475b2..bd61205 100644
--- a/fsspec/implementations/tests/test_dbfs.py
+++ b/fsspec/implementations/tests/test_dbfs.py
@@ -129,11 +129,21 @@ def make_mock_diabetes_ds():
 
 
 @pytest.mark.vcr()
-def test_dbfs_file_listing(dbfsFS):
-    assert "/FileStore" in dbfsFS.ls("/", detail=False)
-    assert {"name": "/FileStore", "size": 0, "type": "directory"} in dbfsFS.ls(
-        "/", detail=True
+def test_dbfs_listing(dbfsFS):
+    dbfsFS.mkdir(
+        "/FileStore/fsspec_dbfs/project_root/empty_dir",
+        create_parents=True,
+        exist_ok=True,
     )
+    assert {
+        "name": "/FileStore/fsspec_dbfs/project_root",
+        "size": 0,
+        "type": "directory",
+    } in dbfsFS.ls("/FileStore/fsspec_dbfs", detail=True)
+    assert dbfsFS.ls("/FileStore/fsspec_dbfs/project_root/empty_dir", detail=True) == []
+    # Test that FileNotFoundError is raised when listing a missing directory
+    with pytest.raises(FileNotFoundError):
+        dbfsFS.ls("/FileStore/fsspec_dbfs/project_root/missing_directory", detail=True)
 
 
 @pytest.mark.vcr()
diff --git a/fsspec/implementations/tests/test_local.py b/fsspec/implementations/tests/test_local.py
index 2e4c07c..fd6b656 100644
--- a/fsspec/implementations/tests/test_local.py
+++ b/fsspec/implementations/tests/test_local.py
@@ -1,4 +1,5 @@
 import bz2
+import errno
 import gzip
 import os
 import os.path
@@ -562,6 +563,25 @@ def test_multiple_filesystems_use_umask_cache(tmpdir):
     assert get_umask.cache_info().hits == 1
 
 
+def test_transaction_cross_device_but_mock_temp_dir_on_wrong_device(tmpdir):
+    # If the temporary file for a transaction is not on the correct device,
+    # os.rename in shutil.move will raise EXDEV and lookup('chmod') will raise
+    # a PermissionError.
+    fs = LocalFileSystem()
+    with (
+        patch(
+            "os.rename",
+            side_effect=OSError(errno.EXDEV, "Invalid cross-device link"),
+        ),
+        patch(
+            "os.chmod",
+            side_effect=PermissionError("Operation not permitted"),
+        ),
+    ):
+        with fs.transaction, fs.open(tmpdir + "/afile", "wb") as f:
+            f.write(b"data")
+
+
 def test_make_path_posix():
     cwd = os.getcwd()
     if WIN:
diff --git a/fsspec/tests/test_generic.py b/fsspec/tests/test_generic.py
index bd9745f..6e2ba04 100644
--- a/fsspec/tests/test_generic.py
+++ b/fsspec/tests/test_generic.py
@@ -48,6 +48,34 @@ def test_cat_async(server):
     assert fs.cat(server.realfile) == data
 
 
+def test_cp_one(server, tmpdir):
+    fsspec.filesystem("http", headers={"give_length": "true", "head_ok": "true"})
+    local = fsspec.filesystem("file")
+    fn = f"file://{tmpdir}/afile"
+
+    fs = fsspec.filesystem("generic", default_method="current")
+
+    fs.copy([server.realfile], [fn])
+    assert local.cat(fn) == data
+    fs.rm(fn)
+    assert not fs.exists(fn)
+
+    fs.copy(server.realfile, fn)
+    assert local.cat(fn) == data
+    fs.rm(fn)
+    assert not fs.exists(fn)
+
+    fs.cp([server.realfile], [fn])
+    assert local.cat(fn) == data
+    fs.rm(fn)
+    assert not fs.exists(fn)
+
+    fs.cp_file(server.realfile, fn)
+    assert local.cat(fn) == data
+    fs.rm(fn)
+    assert not fs.exists(fn)
+
+
 def test_rsync(tmpdir, m):
     from fsspec.generic import GenericFileSystem, rsync