[CHORE] Implement dispatcher for intermediate ops and streaming sink

[colin-ho]

In the streaming physical writes PR, we introduced a dispatcher trait, which allows a blocking sink to define how to distribute work across workers: #2992, i.e. partitioned by hash, round robin, etc.

This PR enables intermediate ops and streaming sinks to use dispatcher as well. In the future, we can add a unordered dispatcher which allows workers to accept input on demand instead of round robin.

Stacked on #3235