GH-3558: Properly close buffers

[Fokko]

Rationale for this change

With updating to a later version of Hadoop, we got the error that there are lingering buffers that are not closed properly. This PR addresses this issue by properly releasing the acquired buffers.

Also, closes some of the readers using try-with-resource pattern.

Initially caught by @pan3793 in #3360 (comment)

Closes #3558
Closes #3356

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

[wgtmac]

It would be good if @gszadovszky could take a look at this as this is related to TrackingByteBufferAllocator.

[wgtmac]

Do we want to reflect this in the PR title?

[Fokko]

This is the underbound of the supported Hadoop version, so I think it is best to set this to 3.3.0 before merging this. Unless @steveloughran has an opinion on a reasonable lower-bound :)

[wgtmac]

This looks like a behavior change?

[Fokko]

Yes, it is, but it is not covered in any tests. WDYT @gszadovszky ?

[Fokko]

I don't think this was working properly since Hadoop did raise exception around not properly closing buffers, while this exception wasn't being thrown. Let me dig a bit further here

[gszadovszky]

Yeah. The whole point of this allocator implementation is to fail if something was not released. It may make sense to actually release the buffers so the tests won't leak, be we need to throw that exception, otherwise this class is pointless.

[gszadovszky]

When I've created this class, I've caught a couple of leaks in parquet-java code with it. Not having that exception thrown during our unit tests means the leaks are fixed not that the monitoring class is not working :)

[steveloughran]

so for all cluster filesystems with vector io, it's fine as is. HDFS doesn't support it and for the cloud stores it's all ranged reads straight into allocated buffers.

I think maybe in hadoop we should just cut the attempt to be clever and merge ranges, and just do the parallel reads. On clusterfs work with Owen O'Malley and claude to do the right thing here.

what would be the perceived penalty of reading the whole file block into one allocated buffer, copying the requested pieces into two separate buffers, and then releasing the larger one (a release function is now returned down after all).

[wgtmac]

This assumes the future returns the exact buffer allocated by our allocator. With local checksum verification enabled, Hadoop can return a sliced buffer, so strict allocators like TrackingByteBufferAllocator or ReusingByteBufferAllocator may fail on release or still report a leak. The new test core-site.xml disables that path globally, so this should either guard against sliced vectored IO or have a test with checksum verification left on. (reviewed by Codex)

[Fokko]

Good catch. You're right that registering the exact buffer returned by the future is unsafe: with fs.file.checksum.verify=true, ChecksumFileSystem returns a sliced view of the allocated buffer (and allocates extra checksum buffers it never releases), so strict allocators like TrackingByteBufferAllocator/ReusingByteBufferAllocator would fail on release or report a leak. Disabling checksum verification globally in the test core-site.xml just masked that.

I've reworked it to guard the path properly instead of relying on the config:

• Added a RecordingByteBufferAllocator decorator that wraps the real allocator during readVectored and records the actual buffers Hadoop allocates. We now release those exact buffers (data + checksum) rather than whatever the futures hand back.
• Removed the global core-site.xml override, so the tests run with checksum verification at Hadoop's default (enabled).
• TestParquetFileWriter already parameterizes over vectored/non-vectored with the strict TrackingByteBufferAllocator; with checksum verification now left on, the [vectored : true] case exercises exactly the sliced-buffer path. I added a comment there documenting that intent.

Verified: TestParquetFileWriter, TestParquetReader, TestColumnIndexFiltering, and TestDataPageChecksums all pass with checksum verification on and strict allocators — no leaks, no double-releases.

[wgtmac]

If this range is registered and a later vectored range times out or fails before rowGroup.setReleaser(builder.releaser), the builder is dropped without closing the releaser. We should close the builder releaser on the failure path, or transfer ownership earlier so partially-read row groups still release their buffers. (reviewed by Codex)

[Fokko]

Agreed — if a later range times out or fails, the exception propagates before rowGroup.setReleaser(builder.releaser) runs, so the already-acquired buffers leak.

Fixed by transferring ownership on the failure path:

• Extracted a readChunkPagesForBlock(...) helper (shared by internalReadRowGroup and internalReadFilteredRowGroup) that closes builder.releaser if anything throws before ownership is transferred to the row group.
• In readVectored, moved the buffer registration into a finally block so a partially-completed vectored read still registers every buffer it acquired, ensuring they're released even when a later range fails.