[data] [streaming] Support a streaming_repartition() operator

[ericl]

In several use cases, it is useful to change the block size of datasets in a streaming way. The current repartition() operator is an all-to-all operator and is incompatible with streaming.

We could implement a general purpose streaming_repartition() operator that supports repartitioning in a few streaming-compatible ways:

• Splitting/coalescing blocks into a certain number of rows
• Splitting/coalescing blocks into a certain in-memory byte size
• Splitting/coalescing blocks into K pieces

This could be implemented as a new PhysicalOperator that implements the online repartitioning. This could also replace the current SplitBlocks mechanism from #36352

[luxunxiansheng]

Suppose I have 20 big size files and I implement a specfic datasource for it. I would like to load the datset by read_datasource with a parallesim ,say , 200. Now I see the splitblocks function to split each block to many smaller blocks. My question is , how does the splitblocks work? It will split a single big file in each row into many many binary parts and then to coalesce them somewhere in the downstream?

[ericl]

SplitBlocks works within the read task to split the read output into multiple smaller pieces. These will remain as smaller individual blocks for the remainder of the computation unless the dataset is explicitly repartitioned.

Ray Data will automatically insert SplitBlocks to ensure the desired/autodetected parallelism is met after a read.