implement distributed collect on node level with doc granularity

yes downstreams are never null