Patzifist opened a new issue, #863:
URL: https://github.com/apache/arrow-go/issues/863
### 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
