Reduce memory usage in writer by freeing unused buffers

[chenyangfb]

Description

Currently ChunkedSliceOutput keep buffers in bufferPool and usedBuffers, it never free those buffer, even with reset(), leads to extra memory usage and OOM.

This PR avoid those extra memory usage by freeing unused buffer in chunk supplier during reset. This behavior is controlled by resetOutputBuffer in OrcWriterOptions, and it's disabled by default.

Example ChunkedSliceOutput behaviour before and after the change
Assume first batch of output is large, it used 3.75M
after writing those output, and before we call reset()
the usedBuffers contains a list of buffer with increasing sizes (assuming min buffer is 256k, max buffer is 1M for illustration purpose)
256k, 512k, 1M, 1M, 1M

The second batch of output is smaller, it will only used 1.75M
after writing those output, and before we call reset()
usedBuffers have 256k, 512k, 1M,
bufferPool have 1M, 1M

before the change
we will keep all 5 buffers, and will never free them
after the change
we will keep the 3 buffer which are used (256k, 512k, 1M) and free up the 2 buffer which are unused (1M, 1M)

In other word, previous behaviour only scale up number of buffers, new behaviour also can scale down number of buffer based on memory usage

Impact

Reduce memory usage in writer.

Test Plan

Tested with Spark workload

Example output showing unused buffer getting freed after the change

Reset unused buffers, used 17 chunks (8387584 bytes), unused 65 chunks (68157440 bytes)
Reset unused buffers, used 82 chunks (76545024 bytes), unused 4 chunks (4194304 bytes)

== RELEASE NOTES ==
General change
* Improve memory usage in writer by feeing unused buffers.

[sdruzkin]

Memory allocation is expensive and we added the pool for this reason. Please add performance benchmark results for a case comparing writing a 500MB file with 5-7 columns with the new option enabled and disabled.

[steveburnett]

Thanks for the release note! Just a nit or two suggested.

== RELEASE NOTES ==

General Changes
* Improve memory usage in writer by freeing unused buffers.

[sdruzkin]

If you want to save on the memory, do you really want to keep the last batch of chunks? I suppose you don't.

[sdruzkin]

This is a tricky place. If you look at the ChunkSupplier.get method you will see that ChunkSupplier scales up chunks from 256 bytes all the way to 16MB when it needs to produce a new chunk. If you reset the pool and start producing new chunks you will eventually end up only with big 16MB new chunks. Which is probably opposite of your goal.

I don't know what is a best strategy here, because it's really depends on the data shapes, but I see a few options:

1. Have chunk supplier that does not have a buffer, and alway produces smaller fixed size chunks. Will perform best in terms of overhead, but will regress for small streams.
2. Have chunk supplier that does not have a buffer, and alway produces scaled up chunks with reset resetting the current size to min size. Good middle ground option.
3. Have chunk supplier that does not have a buffer, and alway produces scaled up chunks with reset resetting the current size to min size AND using smaller max chunk size. Will perform even better in terms of overhead, won't regress as much as 1 for small streams.

[prestodb-ci]

Saved that user @chenyangfb is from Meta

[chenyangfb]

Discussed with Sergii offline, ran Validation Service (Vader), results looks good, ~3.5k successful samples without failures https://fburl.com/scuba/dwrf_reader_checksum/crm55ef8