do not call compute from executor nodes

src/array/darray.jl

@@ -342,8 +342,20 @@ function stage(ctx, d::Distribute)
         if d.domainchunks == domainchunks(x)
             return x # already properly distributed
         end
+        Nd = ndims(x)
+        T = eltype(d.data)
+        concat = x.concat
         cs = map(d.domainchunks) do idx
-            delayed(collect)(x[idx])
+            chunks = cached_stage(ctx, x[idx]).chunks
+            shape = size(chunks)
+            (delayed() do shape, parts...
+                if prod(shape) == 0
+                    return Array{T}(shape)
+                end
+                dimcatfuncs = [(x...) -> concat(i, x...) for i in 1:length(shape)]
+                ps = reshape(Any[parts...], shape)
+                collect(treereduce_nd(dimcatfuncs, ps))
+            end)(shape, chunks...)
         end
     else
         cs = map(c -> delayed(identity)(d.data[c]), d.domainchunks)

src/thunk.jl

@@ -27,7 +27,7 @@ mutable struct Thunk
                    persist::Bool=false,
                    cache::Bool=false,
                    cache_ref::Nullable{Any}=Nullable{Any}(),
-                   affinity=Nullable(),
+                   affinity=Nullable()
                   )
         thunk = new(f,xs,id,get_result,meta,persist, cache, cache_ref, affinity)
         _thunk_dict[id] = thunk

test/array.jl

@@ -53,6 +53,9 @@ end
     @test Distribute(Blocks(1,1), x) == x
     #test_dist(rand(100, 100))
     #test_dist(sprand(100, 100, 0.1))
+
+    x = distribute(rand(10), 2)
+    @test collect(distribute(x, 3)) == collect(x)
 end
 
 @testset "transpose" begin