feat: add support for Delayed in partitionwise layer creation.

[douglasdavis]

Broadcasting will occur equivalently at each partition because the Delayed object is not partitioned (as shown in the test added in this PR).

[codecov-commenter]

Codecov Report

Attention: 1 lines in your changes are missing coverage. Please review.

Comparison is base (df0ef57) 93.09% compared to head (5c0b00e) 93.10%.
Report is 1 commits behind head on main.

Files	Patch %	Lines
src/dask_awkward/lib/core.py	87.50%	1 Missing ⚠️

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #449      +/-   ##
==========================================
+ Coverage   93.09%   93.10%   +0.01%     
==========================================
  Files          23       23              
  Lines        3272     3279       +7     
==========================================
+ Hits         3046     3053       +7     
  Misses        226      226              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[lgray]

Cool, I'll give it a try in the morning!

[lgray]

I'm getting an odd error when I try to print out the dask awkward array that's made from something using delayed:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/lgray/coffea-dev/coffea/tests/test_lookup_tools.py", line 177, in test_correctionlib
    print(test_out_dak)
  File "/Users/lgray/coffea-dev/dask-awkward/src/dask_awkward/lib/core.py", line 998, in __str__
    f"npartitions={self.npartitions}"
  File "/Users/lgray/coffea-dev/dask-awkward/src/dask_awkward/lib/core.py", line 1083, in npartitions
    return len(self.divisions) - 1
  File "/Users/lgray/miniforge3/envs/ewkcoffea-debug/lib/python3.10/site-packages/dask/delayed.py", line 635, in __len__
    raise TypeError("Delayed objects of unspecified length have no len()")
TypeError: Delayed objects of unspecified length have no len()

I haven't traced it to understand why the output dak array things it's a dask.delayed?

[agoose77]

It's probably that something's constructed Array with the divisions argument taken from delayed.divisions, i.e. in a previous frame.

[douglasdavis]

Is it possible to get a repro for that TypeError?

divisions should always be a tuple (specifically tuple[int, ...] | tuple[None, ...]) on a dak.Array

[lgray]

I'll see if I can get one - I'm trying to make sure first I'm not bamboozling myself.

[lgray]

somehow array.divisions has, itself, become a delayed object!?

[douglasdavis]

Just to jot this information down in a more permanent place: we discovered during slack conversation that if a Delayed object was the first thing passed to map_partitions, we broke an assumption in the code that the first argument would be a dak.Array. The map_partitions implementation is now invariant w.r.t. order of function arguments, but at least one dak.Array is required.

[martindurant]

we broke an assumption in the code that the first argument would be a dak.Array

easy to fix, right?

[douglasdavis]

Yes! fixed in 9c6a65b