Introduce parallelShards() in Streaming Expressions

parallel() uses hash filter partitioning, which doesn't work in some edge cases with high cardinality facets since they kill coordinator on merge phase.
I propose to introduce parallelShards() which will accepts a collection, and spawns per-shard substreams (I'm not sure wether to use distrib=false or shards=foo). So, far it's not clear whether workerCollection is useful for it at all.