Simplify Distributed Stream local intermediate operations

Currently whenever a local intermediate operation (sort, limit, distinct, skip) is invoked on a distributed stream it tries to separate the invoctions based on if it should run remotely or locally. This code is quite hard to read and maintain. It also prevents the ability from adding new functionality without complicating the code substantially.

We can rewrite this to have a new IntermediateDistributedCache that acts a bridge between the distributed (remote) stream and the local stream required for these operations.