[Datasets] Arrow data buffers aren't truncated when pickling zero-copy slice views, leading to huge serialization bloat

[clarkzinzow]

This is a duplicating issue of #22310, so we have a clean issue on which to summarize the problem.

When pickling Arrow data (tables, in our case), Arrow currently doesn't truncate the data buffers for zero-copy slice views of the data (see Arrow ticket). For many of our workloads, this results is sending a potentially very large data buffer over the network for a very small slice. For pathological cases like shuffling, where we might chunk a data block into 1000 chunks and send each to a different reduce, this would involve sending 1000 copies of the entire data block; for a 1 GiB data block, we'd send an aggregate 1 TiB of data over the network instead of the expected 1 GiB. This makes Ray + Arrow essentially unusable.

This bug exists in Arrow 6 through Arrow 10; we worked around the bug in Arrow 6 via explicit copying when slicing data, e.g. in our shuffle implementation, but Arrow 7+ adds internal zero-copy slicing that makes this workaround coverage untenable (e.g. when reading Parquet).

Solution

We can bypass this buggy pickle path by registering a custom serializer for Arrow data that properly truncates these buffers.

[pcmoritz]

Is it possible to fix the pickling path upstream and truncate the buffers there?

[clarkzinzow]

@pcmoritz Yes, I'm planning on doing that for Arrow 11: https://issues.apache.org/jira/browse/ARROW-10739