implement distributed collect on node level with doc granularity