PERF: get all partition widths/lengths in parallel instead of serially.

[mvashishtha]

The reproducing script is the same as in #4493, but the solution here is different: instead of getting all lengths/widths serially, we should do so in parallel. We will need to add a new method at the physical layer for that.

This solution will not save us the cost of serializing the call queue drain result after computing length, but it will let us get all the lengths/widths at once instead of serially.

[noloerino]

I see this is already implemented for Dask by #4420. Is adding a similar implementation for PandasOnRay sufficient, or do we also need similar changes for PyarrowOnRay and OmnisciOnNative as well?

[mvashishtha]

@noloerino here are all the places I can find where we are getting uncached partition shapes serially. I found them by searching for width() in the Modin codebase.

• PandasDataFrame._copartition for both dask and ray:
• PandasDataFrame. _row_lengths for ray: note that dask is using its parallel implementation
• PandasDataFrame._column_widths for ray note that dask dataframe is using its parallel implementation
• in partition manager for both dask and ray
• within dask virtual partitions
• within ray virtual partitions

We do get lengths serially in rebalance_partitions, but that's okay because we check first that the lengths are cached.

[mvashishtha]

It's hard to find a case where this optimization is useful. See my comment here: #4683 (comment)

Given that this optimization doesn't seem to give any major gains, it doesn't seem to be worth the extra code complexity in #4683. I'll close this issue for now.