Implement lazy Array.__iter__ (#7905)

This doesn't fix the original issue pandas-dev/pandas#38645, but hopefully it'll make it easier for pandas to know that it should sanitize dask.arrays.

dask/array/core.py

@@ -1502,6 +1502,10 @@ def name(self, val):
             "please set ``._name``"
         )
 
+    def __iter__(self):
+        for i in range(len(self)):
+            yield self[i]
+
     __array_priority__ = 11  # higher than numpy.ndarray and numpy.matrix
 
     def __array__(self, dtype=None, **kwargs):

dask/array/tests/test_array_core.py

@@ -4210,6 +4210,21 @@ def test_normalize_chunks_nan():
     assert "auto" in str(info.value)
 
 
+def test_pandas_from_dask_array():
+    pd = pytest.importorskip("pandas")
+    from dask.dataframe._compat import PANDAS_GT_130, PANDAS_GT_131
+
+    a = da.ones((12,), chunks=4)
+    s = pd.Series(a, index=range(12))
+
+    if PANDAS_GT_130 and not PANDAS_GT_131:
+        # https://github.com/pandas-dev/pandas/issues/38645
+        assert s.dtype != a.dtype
+    else:
+        assert s.dtype == a.dtype
+        assert_eq(s.values, a)
+
+
 def test_from_zarr_unique_name():
     zarr = pytest.importorskip("zarr")
     a = zarr.array([1, 2, 3])

dask/dataframe/_compat.py

@@ -11,6 +11,7 @@
 PANDAS_GT_120 = PANDAS_VERSION >= parse_version("1.2.0")
 PANDAS_GT_121 = PANDAS_VERSION >= parse_version("1.2.1")
 PANDAS_GT_130 = PANDAS_VERSION >= parse_version("1.3.0")
+PANDAS_GT_131 = PANDAS_VERSION >= parse_version("1.3.1")
 
 
 if PANDAS_GT_100: