Consider tree reduction for barrier

In our shuffle, all transfer tasks reduce to a single barrier task. This is fine because the data we're submitting is trivial. However, this requires the worker which is executing the barrier to connect to all workers and fetch their results. It can only start fetching this data once the barrier is completed, i.e. during this time the cluster will effectively idle


This can be easily seen in these kinds of task streams (This is TPCH Q5) where the large gaps are the barriers.

If we applied a tree reduction, most of the results could already be pre-fetched to few workers to reduce this gap. The reduction could already verify uniqueness and restart tasks if necessary but this wouldn't even be required. It could deduplicate, though.