combine streaming and distributed in copy into

[youngsofun]

Summary

sketch:

//path_stream = list(dirs) 
while paths = path_stream.collect(n):
  splits = split(paths)
  for (part, node) in distribute(cluster.nodes, splits)
      node.fragment_exchanges.try_send(part) 

some considerations:

1. reuse fragment_exchanges to send parts
2. reuse static_exchanges too.
3. fragment_exchanges may return detailed progress to help with fair distribution of
4. add a setting enable_distributed_copy
5. get head N of list(dirs), before starting progress, if < N, no need for streaming.

related issues:

distributed copy

• allow executing COPY INTO in a cluster #6395 @zhang2014 @RinChanNOWWW

streaming copy

• Streaming COPY #7889 @Xuanwo