further compact blocks when insert/load data with many threads

[youngsofun]

Summary

already compacted in each threads.

[youngsofun]

compactor----------->SinkBigBlock-------------------+
                                                    |
                                                    |
                                                    v
                                                 compactor--------->SinkFinal
                                                    ^
                                                    |
                                                    |
compactor----------->SinkBigBlock-------------------+

Sink to encode block to parquet and write to s3.

this is already used in unloading to file.
we can extract this as a shared code (Sink as a type parameter)

---

1. For copy into from file & streaming load, per thread compactor can be avoided
2. SinkBigBlock keep a queue of small_blocks (including those split from big ones)
   1. put small blocks here and write big blocks at once,
   2. if current small_blocks is big in total, write them;
   3. when input is finished, send the remaining of small_blocks to SinkFinal.

@sundy-li @zhyass @dantengsky

[youngsofun]

related

1. try avoid compactor for copy into/streaming load with row-based format #7760
2. Feature: Make Port and InputPort/OutputPort/SharedData generic #8353

[youngsofun]

because a processor has only one output, have to choose 1 of the 2:

1. merge compactor into Sink ( call compactor inside)
2. add a flag in the DataBlock indicating whether to SinkBigBlock should process it or pass it to downstream

@sundy-li

[sundy-li]

because a processor has only one output

why not use resize the processor into one ? The output could be Option<Outport>

[youngsofun]

@sundy-li I mean SinkBigBlock do not have information for branching: sink at once or pass it to downstream (compactor and final sink, resized to 1)