fix: multithreaded file source breaks interpretation (#1036)

* test passes

* add skeleton for test

* test different handlers

* download file from data repo fork

* download file from data repo fork

* reproducer test

* revert changes to fsspec source

* aiohttp skip

* test mt source

* rerun on socket failure

* check data length

* remove problematic interpretation

* skip branch if it has sub-branches

* rename test

* pass failing test

* simplified

* check for AsGrouped interpretation instead

* fsspec-xrootd is not a direct Uproot dependency (fsspec itself will be)

* Same result, but more straightforward logic.

* use another test file with similar structure to avoid adding new test file

---------

Co-authored-by: Jim Pivarski <[email protected]>

.github/workflows/build-test.yml

@@ -67,7 +67,7 @@ jobs:
 
       - name: Run pytest
         run: |
-          python -m pytest -vv tests --reruns 3 --reruns-delay 30 --only-rerun "(?i)http|timeout|connection"
+          python -m pytest -vv tests --reruns 3 --reruns-delay 30 --only-rerun "(?i)http|timeout|connection|socket"
 
   vanilla-build:
     strategy:
@@ -91,4 +91,4 @@ jobs:
 
       - name: Run pytest
         run: |
-          python -m pytest -vv tests --reruns 3 --reruns-delay 30 --only-rerun "(?i)http|timeout|connection"
+          python -m pytest -vv tests --reruns 3 --reruns-delay 30 --only-rerun "(?i)http|timeout|connection|socket"

src/uproot/behaviors/TBranch.py

@@ -11,7 +11,6 @@
 objects themselves.
 """
 
-
 import queue
 import re
 import sys
@@ -21,6 +20,7 @@
 import numpy
 
 import uproot
+import uproot.interpretation.grouped
 import uproot.language.python
 from uproot._util import no_filter
 
@@ -1809,7 +1809,10 @@ def get_from_cache(branchname, interpretation):
         checked = set()
         for _, context in expression_context:
             for branch in context["branches"]:
-                if branch.cache_key not in checked:
+                if branch.cache_key not in checked and not isinstance(
+                    branchid_interpretation[branch.cache_key],
+                    uproot.interpretation.grouped.AsGrouped,
+                ):
                     checked.add(branch.cache_key)
                     for (
                         basket_num,

tests/test_0692_fsspec_reading.py

@@ -3,6 +3,9 @@
 import pytest
 import uproot
 import uproot.source.fsspec
+import uproot.source.file
+import uproot.source.xrootd
+import uproot.source.s3
 
 import skhep_testdata
 import queue
@@ -123,6 +126,7 @@ def test_open_fsspec_ssh(handler):
 )
 def test_open_fsspec_xrootd(handler):
     pytest.importorskip("XRootD")
+    pytest.importorskip("fsspec_xrootd")
     with uproot.open(
         "root://eospublic.cern.ch//eos/root-eos/cms_opendata_2012_nanoaod/Run2012B_DoubleMuParked.root",
         handler=handler,
@@ -208,3 +212,27 @@ def test_fsspec_zip(tmp_path):
     ) as branch:
         data = branch.array(library="np")
         assert len(data) == 40
+
+
+# https://github.com/scikit-hep/uproot5/issues/1035
+@pytest.mark.parametrize(
+    "handler",
+    [
+        uproot.source.file.MemmapSource,
+        uproot.source.file.MultithreadedFileSource,
+        uproot.source.fsspec.FSSpecSource,
+        None,
+    ],
+)
+def test_issue_1035(handler):
+    with uproot.open(
+        skhep_testdata.data_path("uproot-issue-798.root"),
+        handler=handler,
+        use_threads=True,
+        num_workers=10,
+    ) as f:
+        for _ in range(25):  # intermittent failure
+            tree = f["CollectionTree"]
+            branch = tree["MuonSpectrometerTrackParticlesAuxDyn.truthParticleLink"]
+            data = branch.array()
+            assert len(data) == 40