fix: dask distributed fsspec issue (#1065)

* add dask distributed to dev dependencies

* work on test for issue 1063

* test multiple handlers

* file handle pickling

* clear file handle before pickling

Author: lobis

pyproject.toml

@@ -53,7 +53,7 @@ requires-python = ">=3.8"
 dev = [
   "boost_histogram>=0.13",
   "dask-awkward>=2023.12.1",
-  "dask[array]",
+  "dask[array,distributed]",
   "hist>=1.2",
   "pandas",
   "awkward-pandas"

src/uproot/source/fsspec.py

@@ -61,6 +61,7 @@ def __repr__(self):
         return f"<{type(self).__name__} {path} at 0x{id(self):012x}>"
 
     def __getstate__(self):
+        self._fh = None
         state = dict(self.__dict__)
         state.pop("_executor")
         state.pop("_file")

tests/test_1063_dask_distributed.py

@@ -0,0 +1,22 @@
+import pytest
+import skhep_testdata
+
+import uproot
+import uproot.source.file
+import uproot.source.fsspec
+
+dask = pytest.importorskip("dask")
+dask_awkward = pytest.importorskip("dask_awkward")
+dask_distributed = pytest.importorskip("dask.distributed")
+
+
+@pytest.mark.parametrize(
+    "handler",
+    [None, uproot.source.file.MemmapSource, uproot.source.fsspec.FSSpecSource],
+)
+def test_issue_1063(handler):
+    file_path = skhep_testdata.data_path("uproot-issue121.root")
+
+    with dask_distributed.Client():
+        events = uproot.dask({file_path: "Events"}, handler=handler)
+        dask.compute(events.Muon_pt)