fix: make dask_property picklable

property keeps fget/fset/fdel in C slots and has no reduction of its own, so
behavior classes defined in __main__ or a notebook could not be sent to a
distributed worker.

Assisted-by: ClaudeCode:claude-opus-4.8

Author: ikrommyd

src/coffea/util.py

@@ -369,6 +369,12 @@ def impl(*args, **kwargs):
     return descriptor
 
 
+def _rebuild_dask_property(fget, fset, fdel, doc, dask_get):
+    prop = _DaskProperty(fget, fset, fdel, doc)
+    prop._dask_get = dask_get
+    return prop
+
+
 class _DaskProperty(property):
     _dask_get = None
 
@@ -377,6 +383,16 @@ def dask(self, func):
         self._dask_get = _make_dask_descriptor(func)
         return self
 
+    def __reduce__(self):
+        # property keeps fget/fset/fdel in C slots and offers no reduction, so
+        # a behavior class carrying a dask_property cannot be pickled by value
+        # (which is what cloudpickle does for classes defined in __main__ or a
+        # notebook) unless we provide one.
+        return (
+            _rebuild_dask_property,
+            (self.fget, self.fset, self.fdel, self.__doc__, self._dask_get),
+        )
+
 
 def _adapt_naive_dask_get(func):
     def wrapper(self, dask_array, *args, **kwargs):

tests/test_util.py

@@ -71,3 +71,41 @@ def output(x):
     finally:
         if os.path.exists(filename):
             os.remove(filename)
+
+
+def test_dask_property_is_picklable():
+    """Behavior classes defined outside an importable module get pickled by
+    value (e.g. by cloudpickle, when a dask graph goes to a distributed worker),
+    which walks the class dict -- and plain property objects cannot be pickled.
+    """
+    cloudpickle = pytest.importorskip("cloudpickle")
+
+    from coffea.util import dask_property
+
+    class Thing:
+        def __init__(self, x):
+            self.x = x
+
+        @dask_property
+        def doubled(self):
+            """twice x"""
+            return 2 * self.x
+
+        @doubled.dask
+        def doubled(self, dask_array):
+            return 20 * dask_array.x
+
+        @dask_property(no_dispatch=True)
+        def tripled(self):
+            return 3 * self.x
+
+    # defined in a function body, so this has to go by value
+    unpickled = cloudpickle.loads(cloudpickle.dumps(Thing))
+
+    assert unpickled(1).doubled == 2
+    assert unpickled(1).tripled == 3
+
+    doubled = unpickled.__dict__["doubled"]
+    assert doubled.__doc__ == "twice x"
+    assert doubled._dask_get(unpickled(1), unpickled, Thing(3)) == 60
+    assert unpickled.__dict__["tripled"]._dask_get(unpickled(2), unpickled, None) == 6