Support dask.bag reductions on no partitions

mrocklin · May 10, 2017 · 05481cc · 05481cc
1 parent 917bbd4
commit 05481cc
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/dask/bag/core.py b/dask/bag/core.py
@@ -743,6 +743,9 @@ def reduction(self, perpartition, aggregate, split_every=None,
             b = c
             depth += 1
 
+        if not self.npartitions:
+            return Item({b: (aggregate, [])}, b)
+
         if out_type is Item:
             dsk[b] = dsk.pop((b, 0))
             return Item(merge(self.dask, dsk), b)

diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
@@ -1142,3 +1142,11 @@ def test_temporary_directory():
         b2 = b.groupby(lambda x: x % 2)
         b2.compute()
         assert any(fn.endswith('.partd') for fn in os.listdir(os.getcwd()))
+
+
+def test_empty_bag():
+    b = db.from_sequence([])
+    assert b.map(inc).all().compute(get=dask.get)
+    assert not b.map(inc).any().compute(get=dask.get)
+    assert not b.map(inc).sum().compute(get=dask.get)
+    assert not b.map(inc).count().compute(get=dask.get)