[data] Log a warning if the batch size is misconfigured in a way that would grossly reduce parallelism for actor pool.

[ericl]

Why are these changes needed?

In a situation such as:

import ray
import time

def slow_model(x):
    time.sleep(1)
    return x

# This will try to batch 4096 elements into 1 task! resulting in no parallelism.
ds = ray.data.range(100).map_batches(
     slow_model, compute=ray.data.ActorPoolStrategy(size=4),
     batch_size=4096)
ds.show()

Print a warning after execution completes that the batch size is too large.

[amogkam]

even if there are a small number of blocks, if the batch size is small, this can still be parallelized across multiple actors?

[ericl]

I don't believe we ever auto split blocks due to batch size (only ever coalesce blocks up to the min batch size if the user explicitly specifies it).

[amogkam]

Oh I see...I thought it was using the same mechanism as for iter_batches