Severe sync.Pool degradation and infinite heap allocations in GetColumnBloomFilter due to pool pollution and ResizeNoShrink(0)

[Patzifist]

Describe the bug, including details regarding any error messages, version, and platform.

The Problem
There is a massive, highly inefficient memory allocation overhead during parallel execution of GetColumnBloomFilter under high concurrency (e.g., streaming thousands of Parquet files via 65 parallel splits).

The root cause lies in a combination of two architectural design flaws in the Parquet reader:

1. Pool Pollution (Variable Buffer Sizes): The reader uses a single, shared r.bufferPool (sync.Pool) for everything. It simultaneously recycles tiny page headers (re-sliced to 256 bytes or 4 KB) and heavy Bloom Filter bitsets (expanded to 2 MB) into the exact same pool.
2. Capacity Destruction: Inside parquet/metadata/cleanup.go, the asynchronous runtime.AddCleanup task (or explicit defers) resets recycled buffers using data.ResizeNoShrink(0).

Reproducer Log / Evidence
We ran a concurrent stress test simulating a standard Trino/Iceberg split engine: 65 concurrent goroutines executing 100 sequential file reads each, processing variable-sized Bloom Filter segments (randomized between 512 KB and 1 MB to mimic real production data).

=== RUN   TestGetColumnBloomFilter_OriginalPoolLeak
    bloom_filter_leak_test.go:116: Original stress test processed 6500 operations in 1.396053644s

Cumulative Allocated Memory: 13024.81 MB
Total OS Mallocs Count:      254009

package metadata

import (
	"bytes"
	"context"
	"fmt"
	"math/rand"
	"runtime"
	"sync"
	"testing"
	"time"

	"github.com/apache/arrow-go/v18/arrow/memory"
	"github.com/apache/arrow-go/v18/parquet"
	format "github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet"
	"github.com/apache/arrow-go/v18/parquet/internal/thrift"
	"github.com/apache/arrow-go/v18/parquet/schema"
)

type fakeReader struct {
	*bytes.Reader
}

func (f *fakeReader) ReadAt(p []byte, off int64) (int, error) {
	return f.Reader.ReadAt(p, off)
}

func TestGetColumnBloomFilter_OriginalPoolLeak(t *testing.T) {
	ctx := context.Background()

	maxSize := 2 * 1024 * 1024

	originalBufferPool := &sync.Pool{
		New: func() any {
			buf := memory.NewResizableBuffer(memory.NewGoAllocator())
			runtime.SetFinalizer(buf, func(obj *memory.Buffer) {
				obj.Release()
			})
			return buf
		},
	}

	runtime.GC()
	var memStart runtime.MemStats
	runtime.ReadMemStats(&memStart)

	const (
		goroutines          = 65
		iterationsPerWorker = 100
	)

	var wg sync.WaitGroup
	startSignal := make(chan struct{})

	for range goroutines {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-startSignal

			for range iterationsPerWorker {
				bloomFilterDataSize := int32(512*1024 + rand.Intn(512*1024))
				bloomFilterReadSize := int32(4096)

				header := format.BloomFilterHeader{
					NumBytes:    bloomFilterDataSize,
					Algorithm:   &defaultAlgorithm,
					Hash:        &defaultHashStrategy,
					Compression: &defaultCompression,
				}

				serializer := thrift.NewThriftSerializer()
				headerBytes, _ := serializer.Write(ctx, &header)

				localFileData := make([]byte, maxSize)
				copy(localFileData, headerBytes)

				var offset int64 = 0
				columnMetaData := format.ColumnMetaData{
					Type:              format.Type_BYTE_ARRAY,
					PathInSchema:      []string{"test_col"},
					Codec:             format.CompressionCodec_UNCOMPRESSED,
					BloomFilterOffset: &offset,
					BloomFilterLength: &bloomFilterReadSize,
				}
				thriftColumnChunk := format.ColumnChunk{MetaData: &columnMetaData}
				thriftRowGroup := format.RowGroup{Columns: []*format.ColumnChunk{&thriftColumnChunk}}

				node, _ := schema.NewPrimitiveNode("test_col", parquet.Repetition(format.FieldRepetitionType_REQUIRED), parquet.Type(format.Type_BYTE_ARRAY), -1, -1)
				rootGroup, _ := schema.NewGroupNode("schema", parquet.Repetition(format.FieldRepetitionType_REPEATED), schema.FieldList{node}, -1)
				sc := schema.NewSchema(rootGroup)

				threadRdr := &RowGroupBloomFilterReader{
					input:          &fakeReader{Reader: bytes.NewReader(localFileData)},
					sourceFileSize: int64(len(localFileData)),
					bufferPool:     originalBufferPool,
					rgMeta:         NewRowGroupMetaData(&thriftRowGroup, sc, nil, nil),
				}

				bf, err := threadRdr.GetColumnBloomFilter(0)
				if err != nil {
					continue
				}

				if b, ok := bf.(*blockSplitBloomFilter); ok && b.data != nil {
					b.data.ResizeNoShrink(0)
					originalBufferPool.Put(b.data)
				}
			}
		}()
	}

	startTime := time.Now()
	close(startSignal)
	wg.Wait()
	t.Logf("Original stress test processed 6500 operations in %v", time.Since(startTime))

	var memEnd runtime.MemStats
	runtime.ReadMemStats(&memEnd)

	totalAllocatedMB := float64(memEnd.TotalAlloc-memStart.TotalAlloc) / 1024 / 1024
	heapInuseMB := float64(memEnd.HeapInuse-memStart.HeapInuse) / 1024 / 1024

	fmt.Printf("Cumulative Allocated Memory: %.2f MB\n", totalAllocatedMB)
	fmt.Printf("Active Heap In Use Memory:   %.2f MB\n", heapInuseMB)
	fmt.Printf("Total OS Mallocs Count:      %d\n\n", memEnd.Mallocs-memStart.Mallocs)

	const MaxAllowedAllocMB = 500.0
	if totalAllocatedMB > MaxAllowedAllocMB {
		t.Errorf("FAIL DETECTED: sync.Pool memory leak verified! Allocated %.2f MB", totalAllocatedMB)
	}
}

Component(s)

Parquet

[zeroshade]

Capacity Destruction: Inside parquet/metadata/cleanup.go, the asynchronous runtime.AddCleanup task (or explicit defers) resets recycled buffers using data.ResizeNoShrink(0).

ResizeNoShrink(0) doesn't reset the capacity of the buffers, it explicitly maintains the capacity and resets only the length. Resizing the buffer after the ResizeNoShrink call will not perform any realloc. The pool is recycled correctly.

Cumulative Allocated Memory: 13024.81 MB

Isn't this caused by your own loop with the make inside of it?

				localFileData := make([]byte, maxSize)
				copy(localFileData, headerBytes)

65 * 100 * 2 MB = 13,000 MB

So the cumulative memory allocation in your test is actually your own make in each iteration, not the sync.Pool.

Pool Pollution (Variable Buffer Sizes): The reader uses a single, shared r.bufferPool (sync.Pool) for everything. It simultaneously recycles tiny page headers (re-sliced to 256 bytes or 4 KB) and heavy Bloom Filter bitsets (expanded to 2 MB) into the exact same pool.

This part is a legitimate concern and issue that should probably be fixed.

[zeroshade]

I tried to reproduce this myself and the test you've provided here doesn't actually exercise the bloom filter pool.

Since your test sets BloomFilterOffset = 0, GetColumnBloomFilter returns nil, nil immediately before reading the header, allocating the bitset, touching the pool or registering cleanup. It just returns the raw deserialized bytes from thrift.

It's also important to remember that TotalAlloc from runtime.MemStats is measuring cumulative allocation (churn) not retained memory, so it's not evidence of a leak on its own.

If I set a non-zero offset so the read path runs against the pool (Size() returns the expected 1MB), then the pool reuses the bitset buffer and we get ~18 allocs/op and ~2.7 KB/op if we promptly return the buffer, the 1MB buffer is not reallocated per call. ResizeNoShrink(0) preserves the capacity; only Resize(0) frees the memory. Recreating the RowGroupBloomFilterReader each iteration doesn't change this as long as the pool is shared.

There is a measurable effect that can be improved here due to the runtime.AddCleanup though. Because cleanup is GC-timing dependant, under a burst, the buffers being returned to the pool will lag and we'll see cumulative allocations go up (depending on the value of GOGC it could be anywhere from 1.5GB - 4GB from my testing), all eventually reclaimed by the GC.

Ultimately, this is a churn/latency problem, not a failure of the pool. There's definitely room for us to improve and more promptly return the buffer to the pool though.