Optimize native shuffle to write schema once per output partition instead of once per batch

[andygrove]

What is the problem the feature request solves?

Native shuffle currently encodes the name of the compression codec and the IPC schema once per batch. Storing the schema per batch was originally a requirement because the schema could vary between batches due to dictionary encoding but this is no longer the case.

Describe the potential solution

Write the compression codec name and schema once per partittion. This can be implemented in ShuffleBlockWriter::try_new. Update ShuffleBlockWriter::write_batch to no longer write the header per batch.

Make corresponding changes in the reader NativeBlockDecoderIterator.scala.

We need to make sure that this approach works correctly with spilling as well.

Additional context

No response

[andygrove]

I ran some experiments and I no longer think that this is worth doing. For batches with a tiny schema (e.g. single column) I saw a ~25% reduction in size when re-using a single encoder and stream writer across all batches, but this quickly dropped to ~1% once using batches with 100 columns of mixed types.

[andygrove]

There could still be some CPU savings by writing the schema less often, even if it does not help much with file size.