You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "Patzifist (via GitHub)" <gi...@apache.org> on 2026/06/25 12:58:57 UTC
[PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Patzifist opened a new pull request, #864:
URL: https://github.com/apache/arrow-go/pull/864
… GetColumnBloomFilter
### Rationale for this change
Under high concurrency (65 parallel workers), `GetColumnBloomFilter` completely degrades `sync.Pool`. Calling `ResizeNoShrink(0)` drops the internal buffer capacity to 0, forcing a fresh heap allocation on every single iteration and churning 13 GB of garbage in 1.4 seconds.
### What changes are included in this PR?
Fixes #863
1. Fixed buffer recycling by using data.Reset(data.Buf()[:cap(data.Buf())]) instead of ResizeNoShrink(0) to preserve backing slice capacity.
2. Isolated heavy Bloom filter allocations from small metadata headers to eliminate pool pollution.
### Are these changes tested?
Yes, added a concurrent stress test suite in bloom_filter_leak_test.go
- Original code (Bug): Allocated: 13024.81 MB (Pool failed completely).
- Fixed code (sync.Pool): Allocated: 66.25 MB (100% memory reuse verified)
### Are there any user-facing changes?
No. This is strictly an internal performance and memory footprint optimization.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4848200385
Hi Zeroshade,
Currently, the upstream implementation of Parquet Bloom Filter reader eats up all available RAM on the server just for Bloom Filter metadata, leading to OOM panics under extreme concurrency.
I've designed a patch that solves this issue by bringing memory allocations down to a strictly bound steady-state, allowing us to safely repurpose that RAM for processing actual data page allocations and Row Group data instead.
Benchmark Results (64 Workers, 100 Iterations, Random NDV Sizes):
To validate the patch, I've written a concurrent benchmark mimicking varying filter sizes and heavy thread interleaving:
BenchmarkGetColumnBloomFilter_OriginalSyncPool-12 1 1176984718 ns/op 6280746824 B/op 233435 allocs/op
BenchmarkGetColumnBloomFilter_ChannelsPatch-12 1 1461299865 ns/op 705041384 B/op 209192 allocs/op
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3554372419
##########
parquet/metadata/bloom_filter_test.go:
##########
@@ -721,3 +723,23 @@ func TestGetSpacedHashesFromBitmap(t *testing.T) {
assert.Equal(t, int(numValid), trueCount+falseCount, "all hashes should be true or false")
})
}
+
+func TestRecycleEncryptedPathPool(t *testing.T) {
+ pool := &sync.Pool{New: func() any { return memory.NewResizableBuffer(memory.NewGoAllocator()) }}
+ r := &RowGroupBloomFilterReader{bufferPool: pool}
+
+ bitset := make([]byte, 1024)
+ bf := &blockSplitBloomFilter{
+ data: memory.NewBufferBytes(bitset),
+ bitset32: arrow.Uint32Traits.CastFromBytes(bitset),
+ }
+
+ r.recycle(bf)
+
+ got := pool.Get().(*memory.Buffer)
+ if got.Mutable() {
Review Comment:
Non-blocking: this guard is non-deterministic. `sync.Pool.Get()` may hand back a freshly `New()`-ed (mutable) buffer even if `recycle` had wrongly pooled the nil-allocator one, in which case this early-`return`s and the test passes without ever hitting the `ResizeNoShrink` panic path. With the current (correct) code it always passes, so as a regression guard it's weak. Consider asserting deterministically that `recycle` did not enqueue the encrypted buffer — e.g. recycle into a freshly-created empty pool and assert the buffer it holds is the allocator-backed one (`Mutable()`), or compare the specific pointer.
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,46 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+// VisitColumnBloomFilter invokes fn for the BloomFilter of the column at index i.
+//
+// Lifetime contract: The BloomFilter passed to fn (and its backing bitset)
+// is recycled immediately after fn returns. Callers must not retain, store,
+// or use the BloomFilter or its data outside the scope of the fn function.
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
+ bf, err := r.GetColumnBloomFilter(i)
+ if err != nil || bf == nil {
+ return err
+ }
+
+ defer r.recycle(bf)
+
+ return fn(bf)
+}
+
+func (r *RowGroupBloomFilterReader) recycle(bf BloomFilter) {
+ if bf == nil {
+ return
+ }
+
+ b, ok := bf.(*blockSplitBloomFilter)
+ if !ok {
+ return
+ }
+
+ b.bitset32 = nil
+
+ if r.bufferPool != nil && b.cancelCleanup != nil {
+ // Stop the GC cleanup so it can't return b.data to the pool a second time.
+ b.cancelCleanup()
Review Comment:
This resolves the double-`Put` blocker from my last round — verified it unregisters in both build variants (`c.Stop` on go1.24, `runtime.SetFinalizer(bf, nil)` on ≤1.23), so the buffer can't be returned to the pool a second time.
Non-blocking: there's still no dedicated regression test for this path. `TestRecycleEncryptedPathPool` only exercises the encrypted (nil-`cancelCleanup`) branch, not the pool-backed one. A test that reads/recycles a *pool-backed* filter, forces `runtime.GC()`, and asserts the same buffer isn't enqueued twice (e.g. two `pool.Get()`s return distinct pointers) would lock this in — the original bug was CI-invisible since it only fires after a GC and the suite doesn't run this under `-race`.
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,46 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+// VisitColumnBloomFilter invokes fn for the BloomFilter of the column at index i.
+//
+// Lifetime contract: The BloomFilter passed to fn (and its backing bitset)
+// is recycled immediately after fn returns. Callers must not retain, store,
+// or use the BloomFilter or its data outside the scope of the fn function.
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
Review Comment:
Nice scoped accessor — the lifetime contract is clear and `recycle` nils `bitset32`/`data` so misuse fails fast. Two non-blocking follow-up notes:
1. Nothing in the library calls `VisitColumnBloomFilter` yet, so the prompt-return win only lands for callers who opt in; `GetColumnBloomFilter` users still rely on GC recycling. Fine if that's the intent.
2. The bloom reader still shares `file.Reader.bufferPool` with the column/record readers, so MB-sized bitsets and small page buffers churn the same pool. A dedicated per-reader bloom pool would resolve that pollution — good as a follow-up, not needed here.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3507538694
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,34 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
+ bf, err := r.GetColumnBloomFilter(i)
+ if err != nil || bf == nil {
+ return err
+ }
+
+ defer r.recycle(bf)
+
+ return fn(bf)
+}
+
+func (r *RowGroupBloomFilterReader) recycle(bf BloomFilter) {
+ if b, ok := bf.(*blockSplitBloomFilter); ok && b.data != nil {
+ if b.cancelCleanup != nil {
+ b.cancelCleanup()
+ }
+
+ b.data.Reset(b.data.Buf()[:cap(b.data.Buf())])
+
+ b.bitset32 = nil
+
+ if r.bufferPool != nil {
+ r.bufferPool.Put(b.data)
Review Comment:
**[blocker] This returns the encrypted-path buffer to the pool.**
`recycle` unconditionally `Put`s `b.data` back into `r.bufferPool`, but the encrypted branch of `GetColumnBloomFilter` (above) builds the filter with `data: memory.NewBufferBytes(bitset)` — a buffer with a **nil allocator** and no `cancelCleanup`. So `VisitColumnBloomFilter` on an encrypted column will:
1. push a foreign buffer holding **decrypted plaintext** into the reader's general-purpose buffer pool, and
2. **panic** on the next reuse that grows it (`ResizeNoShrink` → `Reserve` → `b.mem.Allocate` on a nil allocator).
Reproducer — drop into `package metadata` and run `go test -run TestRecycleEncryptedPathPool -v ./parquet/metadata/`:
```go
import (
"sync"
"testing"
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/memory"
)
func TestRecycleEncryptedPathPool(t *testing.T) {
pool := &sync.Pool{New: func() any { return memory.NewResizableBuffer(memory.NewGoAllocator()) }}
r := &RowGroupBloomFilterReader{bufferPool: pool}
// Mirror the encrypted path: data wrapped by NewBufferBytes (nil allocator), no cancelCleanup.
bitset := make([]byte, 1024)
bf := &blockSplitBloomFilter{
data: memory.NewBufferBytes(bitset),
bitset32: arrow.Uint32Traits.CastFromBytes(bitset),
}
r.recycle(bf) // pushes the nil-allocator buffer into the pool
got := pool.Get().(*memory.Buffer)
if got.Mutable() {
t.Skip("pool returned a fresh buffer instead of the recycled one; rerun")
}
got.ResizeNoShrink(2 * 1024 * 1024) // grow -> Reserve -> b.mem.Allocate on nil -> panic
}
```
Output on this branch:
```
--- FAIL: TestRecycleEncryptedPathPool (panic: runtime error: invalid memory address or nil pointer dereference)
```
Suggested fix: only recycle a pool-owned buffer — e.g. gate on `b.cancelCleanup != nil` (the encrypted path never registers a cleanup), or verify the buffer is allocator-backed, or have the encrypted branch `Release()` its buffer instead of being eligible for `recycle`.
##########
parquet/metadata/bloom_filter_benchmark_test.go:
##########
@@ -0,0 +1,157 @@
+package metadata
Review Comment:
**[blocker] Missing the ASF license header.** Every source file in the repo carries the 15-line Apache header, and CI enforces it via `dev/release/run_rat.sh` (Apache RAT) plus the pre-commit hook — this file will fail both. Please copy the header from any sibling file (e.g. `bloom_filter_reader_test.go`).
##########
parquet/metadata/bloom_filter_benchmark_test.go:
##########
@@ -0,0 +1,157 @@
+package metadata
+
+import (
+ "bytes"
+ "context"
+ "math/rand"
+ "sync"
+ "testing"
+
+ "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)
+}
+
+type testHarness struct {
+ payload []byte
+ sourceFileSize int64
+ metaList []*RowGroupMetaData
+ metaMask int
+}
+
+func prepareTestHarness(b *testing.B) *testHarness {
+ ctx := context.Background()
+ const maxSize = 4 * 1024 * 1024
+
+ 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)
+
+ const metaCount = 128
+ metaList := make([]*RowGroupMetaData, metaCount)
+ payload := make([]byte, maxSize)
+ var offset int64 = 8
+
+ rnd := rand.New(rand.NewSource(42))
+ for i := 0; i < metaCount; i++ {
+ bloomFilterDataSize := int32(1*1024*1024 + rnd.Intn(200*1024))
+ bloomFilterReadSize := bloomFilterDataSize + 4096
+
+ header := format.BloomFilterHeader{
+ NumBytes: bloomFilterDataSize,
+ Algorithm: &defaultAlgorithm,
+ Hash: &defaultHashStrategy,
+ Compression: &defaultCompression,
+ }
+
+ serializer := thrift.NewThriftSerializer()
+ headerBytes, _ := serializer.Write(ctx, &header)
+ copy(payload[offset:], headerBytes)
Review Comment:
Minor benchmark-fidelity note: `offset` is a single variable fixed at 8 for the whole loop (and every `columnMetaData.BloomFilterOffset` takes `&offset`), so each iteration copies its header over the previous one at the same position. At run time every `metaList[k]` read deserializes the *last* header written here, i.e. one fixed filter size — the 128 "random NDV size" metas don't actually vary the bytes read. If the intent is to exercise varying sizes, give each filter its own offset in `payload` and point that meta's `BloomFilterOffset` at it.
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,34 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
Review Comment:
Please add a doc comment stating the lifetime contract: the `BloomFilter` passed to `fn` (and its backing bitset) is recycled the moment `fn` returns, so callers must not retain or use it afterward. It's good that `recycle` nils `bitset32`/`data` so misuse fails fast rather than reading a recycled buffer — but for a public method that contract should be explicit.
##########
parquet/metadata/bloom_filter_benchmark_test.go:
##########
@@ -0,0 +1,157 @@
+package metadata
+
+import (
+ "bytes"
+ "context"
+ "math/rand"
+ "sync"
+ "testing"
+
+ "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)
+}
+
+type testHarness struct {
+ payload []byte
+ sourceFileSize int64
+ metaList []*RowGroupMetaData
+ metaMask int
+}
+
+func prepareTestHarness(b *testing.B) *testHarness {
+ ctx := context.Background()
+ const maxSize = 4 * 1024 * 1024
+
+ 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)
+
+ const metaCount = 128
+ metaList := make([]*RowGroupMetaData, metaCount)
+ payload := make([]byte, maxSize)
+ var offset int64 = 8
+
+ rnd := rand.New(rand.NewSource(42))
+ for i := 0; i < metaCount; i++ {
+ bloomFilterDataSize := int32(1*1024*1024 + rnd.Intn(200*1024))
+ bloomFilterReadSize := bloomFilterDataSize + 4096
+
+ header := format.BloomFilterHeader{
+ NumBytes: bloomFilterDataSize,
+ Algorithm: &defaultAlgorithm,
+ Hash: &defaultHashStrategy,
+ Compression: &defaultCompression,
+ }
+
+ serializer := thrift.NewThriftSerializer()
+ headerBytes, _ := serializer.Write(ctx, &header)
+ copy(payload[offset:], headerBytes)
+
+ 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}}
+ metaList[i] = NewRowGroupMetaData(&thriftRowGroup, sc, nil, nil)
+ }
+
+ return &testHarness{
+ payload: payload,
+ sourceFileSize: int64(len(payload)),
+ metaList: metaList,
+ metaMask: metaCount - 1,
+ }
+}
+
+func BenchmarkVisitColumnBloomFilter_SyncPool(b *testing.B) {
+ h := prepareTestHarness(b)
+
+ originalArrowPool := &sync.Pool{
+ New: func() any {
+ return memory.NewResizableBuffer(memory.NewGoAllocator())
+ },
+ }
+
+ b.ResetTimer()
+ b.ReportAllocs()
+
+ b.RunParallel(func(pb *testing.PB) {
+ threadRdr := &RowGroupBloomFilterReader{
+ input: &fakeReader{Reader: bytes.NewReader(h.payload)},
+ sourceFileSize: h.sourceFileSize,
+ bufferPool: originalArrowPool,
+ }
+
+ seq := rand.Intn(h.metaMask)
+
+ for pb.Next() {
+ threadRdr.rgMeta = h.metaList[seq&h.metaMask]
+ seq++
+
+ err := threadRdr.VisitColumnBloomFilter(0, func(bf BloomFilter) error {
+ if bf == nil || bf.Size() <= 0 {
+ b.Fatal("bloom filter read path did not run or returned empty bitset")
+ }
+ _ = bf.Hasher()
+ return nil
+ })
+ if err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+}
+
+/*
Review Comment:
Please remove this commented-out `_Channels` benchmark. It references APIs that were dropped from this PR (`parquet.NewBloomFilterPools`, `WithSharedBloomFilterPools`), so it's dead code that can't compile if uncommented.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4929091657
This is going to need a rebase due to #892 getting merged.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4801801399
Ultimately, there is a measurable thing to fix here regarding the latency and churn of memory and how promptly we return the buffer to the pool. But we should improve that without breaking the public `BloomFilter` api, bypassing the reader's allocator, or introducing new process-global pools instead of the per-reader pool.
For reference, here's the test file I used:
```go
const (
reproGoroutines = 65
reproIters = 100
reproOps = reproGoroutines * reproIters
reproBitsetSize = 1 * 1024 * 1024
reproScratch = 2 * 1024 * 1024
reproReadSize = int32(4096)
)
func reproHeaderBytes(t testing.TB) []byte {
t.Helper()
hb, err := thrift.NewThriftSerializer().Write(context.Background(), &format.BloomFilterHeader{
NumBytes: int32(reproBitsetSize),
Algorithm: &defaultAlgorithm,
Hash: &defaultHashStrategy,
Compression: &defaultCompression,
})
if err != nil {
t.Fatalf("serialize bloom filter header: %v", err)
}
return hb
}
// reproFileData lays out [8 byte prefix][thrift header][bitset] so a non-zero
// BloomFilterOffset points at the header.
func reproFileData(t testing.TB) []byte {
t.Helper()
fd := append(make([]byte, 8), reproHeaderBytes(t)...)
return append(fd, make([]byte, reproBitsetSize)...)
}
func reproMeta(t testing.TB, bloomFilterOffset int64) *RowGroupMetaData {
t.Helper()
length := reproReadSize
cmd := format.ColumnMetaData{
Type: format.Type_BYTE_ARRAY,
Encodings: []format.Encoding{format.Encoding_PLAIN},
PathInSchema: []string{"test_col"},
Codec: format.CompressionCodec_UNCOMPRESSED,
BloomFilterOffset: &bloomFilterOffset,
BloomFilterLength: &length,
}
rg := format.RowGroup{Columns: []*format.ColumnChunk{{MetaData: &cmd}}, TotalByteSize: 100, NumRows: 100}
node, err := schema.NewPrimitiveNode("test_col", parquet.Repetition(format.FieldRepetitionType_REQUIRED),
parquet.Type(format.Type_BYTE_ARRAY), -1, -1)
if err != nil {
t.Fatalf("create primitive node: %v", err)
}
root, err := schema.NewGroupNode("schema", parquet.Repetition(format.FieldRepetitionType_REPEATED),
schema.FieldList{node}, -1)
if err != nil {
t.Fatalf("create group node: %v", err)
}
return NewRowGroupMetaData(&rg, schema.NewSchema(root), nil, nil)
}
func reproReader(fileData []byte, meta *RowGroupMetaData, pool *sync.Pool) *RowGroupBloomFilterReader {
return &RowGroupBloomFilterReader{
input: bytes.NewReader(fileData),
sourceFileSize: int64(len(fileData)),
rgMeta: meta,
bufferPool: pool,
}
}
func reproPool() *sync.Pool {
return &sync.Pool{New: func() any {
buf := memory.NewResizableBuffer(memory.NewGoAllocator())
runtime.SetFinalizer(buf, func(o *memory.Buffer) { o.Release() })
return buf
}}
}
// reproRecycle returns the bloom filter's buffer to the pool immediately, doing
// exactly what the reader's runtime.AddCleanup callback does (ResizeNoShrink(0)
// then Put), but synchronously. The registered cleanup is cancelled first so the
// same buffer is not returned twice.
func reproRecycle(bf BloomFilter, pool *sync.Pool) {
b, ok := bf.(*blockSplitBloomFilter)
if !ok || b.data == nil {
return
}
if b.cancelCleanup != nil {
b.cancelCleanup()
}
b.data.ResizeNoShrink(0)
pool.Put(b.data)
}
func reproMeasure(t *testing.T, name string, run func() (read, nilRet int64)) (int64, float64) {
t.Helper()
runtime.GC()
var before runtime.MemStats
runtime.ReadMemStats(&before)
start := time.Now()
read, nilRet := run()
elapsed := time.Since(start)
var after runtime.MemStats
runtime.ReadMemStats(&after)
churnMB := float64(after.TotalAlloc-before.TotalAlloc) / 1024 / 1024
t.Logf("%s\n ops=%d filters-read=%d nil-returns=%d cumulative-alloc(churn)=%.1f MB mallocs=%d elapsed=%v",
name, reproOps, read, nilRet, churnMB, after.Mallocs-before.Mallocs, elapsed)
return read, churnMB
}
func reproRunParallel(work func() BloomFilter) (read, nilRet int64) {
var wg sync.WaitGroup
gate := make(chan struct{})
for range reproGoroutines {
wg.Add(1)
go func() {
defer wg.Done()
<-gate
for range reproIters {
if work() != nil {
atomic.AddInt64(&read, 1)
} else {
atomic.AddInt64(&nilRet, 1)
}
}
}()
}
close(gate)
wg.Wait()
return read, nilRet
}
// Scenario A: the original reproducer. BloomFilterOffset=0 and a fresh 2 MiB file
// buffer allocated per iteration. Allocates ~13 GB while never reading a filter.
func TestBloomFilterPoolRepro_A_OriginalHarness(t *testing.T) {
header := reproHeaderBytes(t)
meta := reproMeta(t, 0)
read, _ := reproMeasure(t, "A original harness: offset=0, fresh 2MiB file per iteration", func() (int64, int64) {
var read, nilRet int64
var wg sync.WaitGroup
gate := make(chan struct{})
for range reproGoroutines {
wg.Add(1)
go func() {
defer wg.Done()
pool := reproPool()
<-gate
for range reproIters {
scratch := make([]byte, reproScratch)
copy(scratch, header)
bf, err := reproReader(scratch, meta, pool).GetColumnBloomFilter(0)
if err != nil || bf == nil {
atomic.AddInt64(&nilRet, 1)
continue
}
atomic.AddInt64(&read, 1)
reproRecycle(bf, pool)
}
}()
}
close(gate)
wg.Wait()
return read, nilRet
})
if read != 0 {
t.Fatalf("expected 0 filters read with offset=0, got %d", read)
}
}
// Scenario B: corrected harness. Non-zero offset (read path runs), shared reader,
// buffer returned to the pool promptly. The 1 MiB bitset buffer is reused.
func TestBloomFilterPoolRepro_B_SharedReaderPromptReturn(t *testing.T) {
fileData := reproFileData(t)
meta := reproMeta(t, 8)
pool := reproPool()
rdr := reproReader(fileData, meta, pool)
bf, err := rdr.GetColumnBloomFilter(0)
if err != nil || bf == nil {
t.Fatalf("sanity read failed: bf=%v err=%v", bf, err)
}
if bf.Size() != int64(reproBitsetSize) {
t.Fatalf("unexpected filter size: got %d want %d", bf.Size(), reproBitsetSize)
}
reproRecycle(bf, pool)
read, churnMB := reproMeasure(t, "B shared reader + prompt return: offset>0", func() (int64, int64) {
return reproRunParallel(func() BloomFilter {
bf, err := rdr.GetColumnBloomFilter(0)
if err != nil || bf == nil {
return nil
}
reproRecycle(bf, pool)
return bf
})
})
if read != reproOps {
t.Fatalf("expected %d filters read, got %d", reproOps, read)
}
if churnMB > 500 {
t.Fatalf("pool not reusing buffers: churn %.1f MB (1 MiB x %d ops would be ~%d MB without reuse)",
churnMB, reproOps, reproOps)
}
}
// Scenario C: corrected harness using the reader's real return path
// (runtime.AddCleanup). GC-timing dependent; reported only.
func TestBloomFilterPoolRepro_C_SharedReaderCleanupOnly(t *testing.T) {
fileData := reproFileData(t)
meta := reproMeta(t, 8)
pool := reproPool()
rdr := reproReader(fileData, meta, pool)
reproMeasure(t, "C shared reader + cleanup-only return: offset>0 (reader's real path)", func() (int64, int64) {
return reproRunParallel(func() BloomFilter {
bf, err := rdr.GetColumnBloomFilter(0)
if err != nil {
return nil
}
return bf
})
})
}
// Scenario D1: a new reader is built every iteration (as in the original harness)
// but the file bytes are shared and the offset is non-zero. A shared pool still
// recycles, so recreating the reader is not what defeated reuse.
func TestBloomFilterPoolRepro_D1_FreshReaderSharedPool(t *testing.T) {
fileData := reproFileData(t)
meta := reproMeta(t, 8)
pool := reproPool()
read, churnMB := reproMeasure(t, "D1 fresh reader + shared pool: offset>0, file shared, prompt return", func() (int64, int64) {
return reproRunParallel(func() BloomFilter {
bf, err := reproReader(fileData, meta, pool).GetColumnBloomFilter(0)
if err != nil || bf == nil {
return nil
}
reproRecycle(bf, pool)
return bf
})
})
if read != reproOps {
t.Fatalf("expected %d filters read, got %d", reproOps, read)
}
if churnMB > 500 {
t.Fatalf("fresh reader defeated shared pool reuse: churn %.1f MB", churnMB)
}
}
// Scenario D2: a new pool is built per read (i.e. a new file.Reader per single
// read). This is the only configuration where per-reader pooling cannot help.
func TestBloomFilterPoolRepro_D2_FreshReaderFreshPool(t *testing.T) {
fileData := reproFileData(t)
meta := reproMeta(t, 8)
reproMeasure(t, "D2 fresh reader + fresh pool per read: offset>0, file shared", func() (int64, int64) {
return reproRunParallel(func() BloomFilter {
bf, err := reproReader(fileData, meta, reproPool()).GetColumnBloomFilter(0)
if err != nil {
return nil
}
return bf
})
})
}
// BenchmarkBloomFilterPoolReuse reports allocs/op for the read path with the
// buffer returned to the pool. A reused 1 MiB bitset does not show up per op.
func BenchmarkBloomFilterPoolReuse(b *testing.B) {
fileData := reproFileData(b)
meta := reproMeta(b, 8)
pool := reproPool()
rdr := reproReader(fileData, meta, pool)
b.ReportAllocs()
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
bf, err := rdr.GetColumnBloomFilter(0)
if err != nil || bf == nil {
continue
}
reproRecycle(bf, pool)
}
})
}
```
Running it locally myself with go test -bench I get the following results:
| Test Case | filters-read | churn |
| -------------- | ---------------- | --------- |
| A - original test (offset = 0) | 0 | 13002.7 MB |
| B - shared reader + prompt return | 6500 | 77.1 MB |
| C - cleanup-only (real path for main) | 6500 | 3419.5 MB |
| D1 - fresh reader + shared pool | 6500 | 39.0 MB |
| D2 - fresh reader + fresh pool | 6500 | 6617.0 MB |
Let's construct a good solution for getting more prompt returns to the pool without breaking the API or defeating the pool. Thoughts?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4848546250
Test for original check
go test -bench=BenchmarkGetColumnBloomFilter_OriginalSyncPool -run=^$ -memprofile=mem.out
goos: linux
goarch: amd64
pkg: github.com/apache/arrow-go/v18/parquet/metadata
cpu: 13th Gen Intel(R) Core(TM) i7-1355U
BenchmarkGetColumnBloomFilter_OriginalSyncPool-12 1 1108550680 ns/op 5550647112 B/op 225956 allocs/op
PASS
ok github.com/apache/arrow-go/v18/parquet/metadata 1.262s
```
package metadata
import (
"bytes"
"context"
"math/rand"
"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 BenchmarkGetColumnBloomFilter_OriginalSyncPool(b *testing.B) {
ctx := context.Background()
const (
workersCount = 64
iterationsPerWorker = 100
maxSize = 4 * 1024 * 1024
)
originalArrowPool := &sync.Pool{
New: func() any {
return memory.NewResizableBuffer(memory.NewGoAllocator())
},
}
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)
payload := make([]byte, maxSize)
b.ResetTimer()
b.ReportAllocs()
for n := 0; n < b.N; n++ {
var wg sync.WaitGroup
startSignal := make(chan struct{})
for w := 0; w < workersCount; w++ {
wg.Add(1)
go func(workerID int) {
defer wg.Done()
<-startSignal
r := rand.New(rand.NewSource(time.Now().UnixNano() + int64(workerID)))
threadRdr := &RowGroupBloomFilterReader{
input: &fakeReader{Reader: bytes.NewReader(payload)},
sourceFileSize: int64(len(payload)),
bufferPool: originalArrowPool,
}
for i := 0; i < iterationsPerWorker; i++ {
bloomFilterDataSize := int32(1*1024*1024 + r.Intn(200*1024))
bloomFilterReadSize := bloomFilterDataSize + 4096
header := format.BloomFilterHeader{
NumBytes: bloomFilterDataSize,
Algorithm: &defaultAlgorithm,
Hash: &defaultHashStrategy,
Compression: &defaultCompression,
}
serializer := thrift.NewThriftSerializer()
headerBytes, _ := serializer.Write(ctx, &header)
var offset int64 = 8
copy(payload[offset:], headerBytes)
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}}
meta := NewRowGroupMetaData(&thriftRowGroup, sc, nil, nil)
threadRdr.rgMeta = meta
bf, err := threadRdr.GetColumnBloomFilter(0)
if err != nil {
return
}
_ = bf
}
}(w)
}
close(startSignal)
wg.Wait()
}
}
```
go tool pprof -http=:8080 mem.out
github.com/apache/arrow-go/v18/parquet/metadata.(*RowGroupBloomFilterReader).GetColumnBloomFilter
/home/alekssmirnov/GolandProjects/arrow-go/parquet/metadata/bloom_filter.go
Total: 1.50MB 5.16GB (flat, cum) 99.56%
490 . . compression: *header.Compression,
491 . . }, nil
492 . . }
493 . .
494 . . headerBuf := r.bufferPool.Get().(*memory.Buffer)
495 . 4.06GB headerBuf.ResizeNoShrink(int(bloomFilterReadSize))
496 . . defer func() {
497 . . if headerBuf != nil {
498 . . headerBuf.ResizeNoShrink(0)
499 . . r.bufferPool.Put(headerBuf)
500 . . }
501 . . }()
502 . .
503 . . if _, err = sectionRdr.Read(headerBuf.Bytes()); err != nil {
504 . . return nil, err
505 . . }
506 . .
507 . 2MB remaining, err := thrift.DeserializeThrift(&header, headerBuf.Bytes())
508 . . if err != nil {
509 . . return nil, err
510 . . }
511 . . headerSize := len(headerBuf.Bytes()) - int(remaining)
512 . .
513 . . if err = validateBloomFilterHeader(&header); err != nil {
514 . . return nil, err
515 . . }
516 . .
517 . . bloomFilterSz := header.NumBytes
518 . . var bitset []byte
519 . . if int(bloomFilterSz)+headerSize <= len(headerBuf.Bytes()) {
520 . . // bloom filter data is entirely contained in the buffer we just read
521 . . bitset = headerBuf.Bytes()[headerSize : headerSize+int(bloomFilterSz)]
522 . . } else {
523 . . buf := r.bufferPool.Get().(*memory.Buffer)
524 . 1.09GB buf.ResizeNoShrink(int(bloomFilterSz))
525 . . filterBytesInHeader := headerBuf.Len() - headerSize
526 . . if filterBytesInHeader > 0 {
527 . . copy(buf.Bytes(), headerBuf.Bytes()[headerSize:])
528 . . }
529 . .
530 . . if _, err = sectionRdr.Read(buf.Bytes()[filterBytesInHeader:]); err != nil {
531 . . return nil, err
532 . . }
533 . . bitset = buf.Bytes()
534 . . headerBuf.ResizeNoShrink(0)
535 . . r.bufferPool.Put(headerBuf)
536 . . headerBuf = buf
537 . . }
538 . .
539 1.50MB 1.50MB bf := &blockSplitBloomFilter{
540 . . data: headerBuf,
541 . . bitset32: arrow.GetData[uint32](bitset),
542 . . hasher: xxhasher{},
543 . . algorithm: *header.Algorithm,
544 . . hashStrategy: *header.Hash,
545 . . compression: *header.Compression,
546 . . }
547 . . headerBuf = nil
548 . 1.50MB addCleanup(bf, r.bufferPool)
549 . . return bf, nil
550 . . }
551 . .
552 . . type FileBloomFilterBuilder struct {
553 . . Schema *schema.Schema
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3521460586
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,44 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+// VisitColumnBloomFilter invokes fn for the BloomFilter of the column at index i.
+//
+// Lifetime contract: The BloomFilter passed to fn (and its backing bitset)
+// is recycled immediately after fn returns. Callers must not retain, store,
+// or use the BloomFilter or its data outside the scope of the fn function.
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
+ bf, err := r.GetColumnBloomFilter(i)
+ if err != nil || bf == nil {
+ return err
+ }
+
+ defer r.recycle(bf)
+
+ return fn(bf)
+}
+
+func (r *RowGroupBloomFilterReader) recycle(bf BloomFilter) {
+ if bf == nil {
+ return
+ }
+
+ b, ok := bf.(*blockSplitBloomFilter)
+ if !ok {
+ return
+ }
+
+ b.bitset32 = nil
+
+ if r.bufferPool != nil && b.cancelCleanup != nil {
+ r.bufferPool.Put(b.data)
+ b.cancelCleanup = nil
Review Comment:
**[blocker] The GC cleanup is never actually cancelled, so `b.data` is returned to the pool twice.**
`addCleanup` (the `//go:build go1.24` variant) registers `runtime.AddCleanup(bf, fn, bf.data)` and stores `bf.cancelCleanup = c.Stop`; `fn` captures the buffer and does `bufferPool.Put(data)`. Here `recycle` sets `b.cancelCleanup = nil`, which only drops the reference to `Stop` — it does **not** unregister the cleanup. So once `VisitColumnBloomFilter` returns and `bf` becomes unreachable, the cleanup still fires and `Put`s the **same** buffer a second time.
- **Go 1.24+**: the buffer is now in the pool twice → two `Get()` callers receive the same `*memory.Buffer` → concurrent use / corruption (the pool-corruption class this PR is meant to fix).
- **Go ≤1.23** (`SetFinalizer` variant): worse — `recycle` already set `b.data = nil`, so the finalizer runs `f.data.ResizeNoShrink(0)` on a nil buffer → panic in the finalizer goroutine.
Fix: invoke the canceler before returning to the pool.
```suggestion
if r.bufferPool != nil && b.cancelCleanup != nil {
// Stop the GC cleanup so it can't return b.data to the pool a second time.
b.cancelCleanup()
b.cancelCleanup = nil
r.bufferPool.Put(b.data)
```
Worth a regression test too: `TestRecycleEncryptedPathPool` covers the encrypted branch, but nothing catches the double-`Put`. A test that recycles a pool-backed filter, forces `runtime.GC()`, and asserts the buffer isn't enqueued twice (e.g. two `pool.Get()`s return distinct pointers) would guard this.
##########
parquet/metadata/bloom_filter.go:
##########
@@ -495,7 +495,7 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
headerBuf.ResizeNoShrink(int(bloomFilterReadSize))
defer func() {
if headerBuf != nil {
- headerBuf.ResizeNoShrink(0)
+ headerBuf.Reset(headerBuf.Buf()[:cap(headerBuf.Buf())])
Review Comment:
Minor / possibly unintended: this changes `headerBuf.ResizeNoShrink(0)` → `Reset(...[:cap])`. `ResizeNoShrink(0)` already retains the backing capacity (it only sets `length = 0`), so this isn't needed for reuse, and it's inconsistent with the cleanup path which still does `data.ResizeNoShrink(0)` before `Put`. Not harmful (`Reset` preserves `mem`/`refCount`), but unless there's a reason I'd revert it to keep the recycling style uniform.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3559410799
##########
parquet/metadata/bloom_filter.go:
##########
@@ -549,6 +551,46 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+// VisitColumnBloomFilter invokes fn for the BloomFilter of the column at index i.
+//
+// Lifetime contract: The BloomFilter passed to fn (and its backing bitset)
+// is recycled immediately after fn returns. Callers must not retain, store,
+// or use the BloomFilter or its data outside the scope of the fn function.
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
Review Comment:
1.Yes, that is exactly the intent. I discovered that when processing massive datasets (e.g., 10+ Terabyte Parquet files), relying solely on GC recycling for Bloom filters causes massive memory spikes and leads to OOM issues because the garbage collector cannot keep up with the allocation rate.I plan to adopt VisitColumnBloomFilter immediately in iceberg-go to explicitly reuse and recycle the buffers. This prompt-return mechanism is essential for keeping the memory footprint stable under heavy production workloads.
2.Agreed, that's a very good point about pool pollution. I will address this by introducing a dedicated per-reader bloom pool in a follow-up PR right after this one gets merged.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4848598340
Thanks for the extra benchmark @Patzifist. Before we iterate on the implementation, I want to make sure we're measuring the right thing — I don't think these numbers show what they appear to.
**What the benchmark is actually measuring**
Both rows ran at `b.N = 1` (the `1` immediately after the benchmark name). The entire 65 × 100 = 6,500-call concurrent stress loop runs *inside a single benchmark iteration*, so the testing framework divides the totals by `N = 1`. That means `B/op` and `allocs/op` are not "per `GetColumnBloomFilter` call" — they're the totals for one 6,500-call batch.
I confirmed this with the `ReproduceProductionLeak` benchmark in the PR by varying `-benchtime`:
```
-benchtime=1x 1 631 MB/op 209135 allocs/op
-benchtime=3x 3 318 MB/op 208677 allocs/op
```
`allocs/op` stays at ~209k regardless of `N`, which confirms the "op" is one whole batch. The real per-filter cost is ~209,000 / 6,500 ≈ **~32 allocs per actual filter read**, not 209k.
Two related points:
- `B/op` is derived from `TotalAlloc`, which is *cumulative* allocation (churn), not *retained* memory. It only ever increases, even as the GC frees everything, so by itself it can't demonstrate a leak or an OOM.
- Taking the two rows at face value, the patched path is **~24% slower** (1.461s vs 1.177s) for **~10% fewer allocations** (209k vs 233k) — the opposite of the trade-off we'd want.
**What I do agree is real**
There is something worth fixing, and it's what we converged on earlier in the thread: today the bitset buffer is only returned to the pool via `runtime.AddCleanup`, which is GC-timing dependent. Under a burst those returns lag and cumulative churn climbs — I measure ~3.1 GB of churn on the cleanup-only path vs ~75 MB when the same buffer is returned to the pool promptly. The pool pollution point (small headers and MB-sized bitsets sharing one pool) is also legitimate. So "return the buffer to the per-reader pool more promptly" is a good goal.
**What would help: a proper reproducer**
To justify a change of this size, could you put together a benchmark that isolates the claim? Specifically:
1. **One op = one `GetColumnBloomFilter` call** — use `b.RunParallel` with `b.ResetTimer()` / `b.ReportAllocs()` so `B/op` and `allocs/op` are genuinely per-call. The scenario B/C harness I posted above is a good starting template.
2. **Use a non-zero `BloomFilterOffset`** so the read path actually runs — assert `Size()` returns the expected byte count. The original repro used `offset = 0`, which returns `nil, nil` before the pool, header, or bitset are ever touched.
3. **Keep allocations that aren't under test out of the measured loop** — in the original harness the `make([]byte, 2 MiB)` per iteration was itself ~13 GB of the churn (6,500 × 2 MiB), unrelated to the pool.
4. **If the claim is OOM / unbounded growth, measure *retained* memory** — `HeapInuse`/`HeapAlloc` sampled over time (or RSS), not `TotalAlloc`. A genuine leak shows up as live memory that doesn't drop after a `runtime.GC()`.
5. **Compare against current `main` on equal footing** — same allocator, and the same buffer-return discipline on both sides (either both rely on cleanup, or both return promptly).
If you can share a repro that shows per-call allocations and/or retained-memory growth on current `main`, that gives us a concrete target to optimize against and to verify any fix. Happy to review the harness once it's posted.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4849168486
Based on your feedback, I have updated the PR to implement **Option A (scoped accessor)**:
1. **Implemented `VisitColumnBloomFilter`**: Added the scoped accessor method with a closure `func(BloomFilter) error`. It uses `defer` to promptly recycle the buffer right after the callback terminates.
2. **Deterministic recycling**: The internal helper cancels the `AddCleanup` to prevent double-returning, shrinks the buffer using `ResizeNoShrink(0)`, and safely returns it to the pool.
3. **Fixed the Benchmark**: Rewrote the benchmark using `b.RunParallel` and proper isolated calls (`b.ResetTimer() / b.ReportAllocs()`) so that `B/op` and `allocs/op` now genuinely reflect a single filter read.
Please take a look when you have a moment!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3475598270
##########
parquet/metadata/bloom_filter_leak_test.go:
##########
@@ -0,0 +1,162 @@
+package metadata
Review Comment:
this is missing the apache license header that is required.
##########
parquet/metadata/bloom_filter.go:
##########
@@ -56,6 +56,11 @@ var (
defaultCompression = format.BloomFilterCompression{UNCOMPRESSED: &format.Uncompressed{}}
)
+var (
+ headerPool = make(chan []byte, 512)
+ filterPool = make(chan *memory.Buffer, 512)
+)
Review Comment:
you're replacing the per-reader, allocator-aware `sync.Pool` with package level globals. You're bypassing the user-configured allocator (which means that if users are using the `mallocator` i.e. CGO/non-go memory, unreleased buffers will leak native memory rather than Go memory that can be reclaimed).
##########
parquet/metadata/bloom_filter.go:
##########
@@ -292,6 +297,13 @@ func (b *blockSplitBloomFilter) WriteTo(w io.Writer, enc encryption.Encryptor) (
return w.Write(b.data.Bytes())
}
+func (b *blockSplitBloomFilter) Close() {
+ if b.cancelCleanup != nil {
+ b.cancelCleanup()
+ b.cancelCleanup = nil
+ }
+}
Review Comment:
this isn't actually called anywhere in the code except for in your new test, so this PR doesn't actually fix anything
##########
parquet/metadata/bloom_filter.go:
##########
@@ -360,6 +372,7 @@ type BloomFilter interface {
Hasher() Hasher
CheckHash(hash uint64) bool
Size() int64
+ Close()
Review Comment:
this is a breaking change that we shouldn't introduce and shouldn't be needed.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): Fix sync.Pool memory leak and capacity destruction in GetColumnBloomFilter [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4848676420
Following up on my last comment with a concrete alternative direction. I measured the original path against a prompt-return path on current `main` (same 64×100 harness you posted, `b.N=1`, go1.26.4), splitting the profile into `alloc_space` (churn) vs `inuse_space` (live after the test's end-of-run GC):
| Path | `alloc_space` (churn) | `inuse_space` (retained) | time/batch |
|---|---|---|---|
| Current `main` (return via `runtime.AddCleanup` only) | ~5.1 GB | ~3.0 GB | 0.49 s |
| Return the bitset buffer to the pool **promptly** | ~0.40 GB | **~83 MB** | 0.088 s |
So the per-reader `sync.Pool` already reuses correctly — the only problem is *when* the buffer goes back. Today the bitset buffer is returned only when the `BloomFilter` is GC'd (`addCleanup` → `runtime.AddCleanup`). Under a burst those cleanups lag, so `Get()` keeps allocating fresh ~1 MB buffers. Returning the same buffer synchronously collapses both churn (~13×) and retained memory (~36×) and runs ~5.5× faster. The remaining ~83 MB is ≈ `concurrency × 1 MB` — one live buffer per in-flight goroutine, which is exactly the bounded steady-state we're after.
The only change from your `BenchmarkGetColumnBloomFilter_OriginalSyncPool` that produces the second row is returning the buffer to the pool instead of `_ = bf`:
```go
bf, err := threadRdr.GetColumnBloomFilter(0)
if err != nil {
return
}
// return the bitset buffer to the pool immediately instead of waiting for GC
if b, ok := bf.(*blockSplitBloomFilter); ok && b.data != nil {
if b.cancelCleanup != nil {
b.cancelCleanup() // cancel the AddCleanup so the buffer isn't returned twice
}
b.data.ResizeNoShrink(0)
originalArrowPool.Put(b.data)
}
```
Same harness, same pool, same allocator — only the return timing changes, and that alone accounts for the 5.1 GB → 0.40 GB / 3.0 GB → 83 MB difference. (There's also a real pool-pollution issue: `file.Reader` hands the bloom reader the *same* `sync.Pool` the column/record readers use, so MB-sized bitsets and small page buffers churn each other.)
## Proposal
Keep `GetColumnBloomFilter` and its `AddCleanup` exactly as they are (back-compat + a GC safety net so nothing leaks if a caller retains the filter), and add an **opt-in prompt-return path**. Two possible shapes — I lean toward (A):
**Option A — scoped accessor (preferred):**
```go
func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
bf, err := r.GetColumnBloomFilter(i)
if err != nil || bf == nil {
return err
}
defer r.recycle(bf) // cancelCleanup() + data.ResizeNoShrink(0) + pool.Put(data)
return fn(bf)
}
```
The buffer returns to the pool the instant the callback ends — deterministic, lexically-scoped lifetime, hard to misuse.
**Option B — explicit recycle:**
```go
bf, err := rg.GetColumnBloomFilter(i)
// ... use bf ...
rg.RecycleBloomFilter(bf) // cancels the cleanup, ResizeNoShrink(0), Put back to pool
```
More flexible for callers who can't wrap usage in a closure, but easier to forget (forgetting just falls back to today's GC behavior, so it's safe, only slower).
Both call the same helper that cancels the registered cleanup (so the buffer isn't double-returned) and puts it back. Either can pair with giving the bloom reader its **own per-reader pool**, separate from the page-buffer pool, to fix the pollution.
## Why this over the current PR
The PR reaches a similar steady-state, but the way it gets there trips each of the constraints I called out earlier:
- **Public API break** — it adds `Close()` to the `BloomFilter` interface; every external implementer/consumer breaks. The proposal adds a method on the *reader* and leaves `BloomFilter` untouched.
- **Allocator bypass** — it reads headers into raw `make([]byte, …)`; the proposal keeps everything in allocator-backed `*memory.Buffer`s.
- **Process-global pools** — it declares package-level `headerPool`/`filterPool` channels (currently unused, but the wrong direction); the proposal keeps strictly per-reader pools.
- **Removes the GC safety net** — it drops `AddCleanup` on the read path, so a filter backed by a CGO/mallocator allocator leaks if the caller never calls `Close()`. The proposal keeps `AddCleanup` as the fallback.
- **Use-after-recycle** — `Close()` resets the buffer to full cap and pushes it into a shared pool while the returned filter's `bitset32` still aliases that memory; a concurrent reader can grab and overwrite it. The scoped accessor bounds the lifetime so that can't happen.
Net: same bounded-memory result, no interface break, no allocator bypass, no global pools, and the existing safety net stays. Happy to put up a PR with Option A + the dedicated pool if that direction works for you.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3525585190
##########
parquet/metadata/bloom_filter.go:
##########
@@ -495,7 +495,7 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
headerBuf.ResizeNoShrink(int(bloomFilterReadSize))
defer func() {
if headerBuf != nil {
- headerBuf.ResizeNoShrink(0)
+ headerBuf.Reset(headerBuf.Buf()[:cap(headerBuf.Buf())])
Review Comment:
Fixed
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "Patzifist (via GitHub)" <gi...@apache.org>.
Patzifist commented on PR #864:
URL: https://github.com/apache/arrow-go/pull/864#issuecomment-4887496171
Thank you for the guidance and the direct patch pointers, `@zeroshade`!
The PR is now fully streamlined and ready for your final review!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #864:
URL: https://github.com/apache/arrow-go/pull/864#discussion_r3560213768
##########
parquet/metadata/bloom_filter.go:
##########
@@ -552,6 +554,46 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i int) (BloomFilter, er
return bf, nil
}
+// VisitColumnBloomFilter invokes fn for the BloomFilter of the column at index i.
+//
+// Lifetime contract: The BloomFilter passed to fn (and its backing bitset)
+// is recycled immediately after fn returns. Callers must not retain, store,
+// or use the BloomFilter or its data outside the scope of the fn function.
+func (r *RowGroupBloomFilterReader) VisitColumnBloomFilter(i int, fn func(BloomFilter) error) error {
+ bf, err := r.GetColumnBloomFilter(i)
+ if err != nil || bf == nil {
+ return err
+ }
+
+ defer r.recycle(bf)
+
+ return fn(bf)
+}
+
+func (r *RowGroupBloomFilterReader) recycle(bf BloomFilter) {
+ if bf == nil {
+ return
+ }
+
+ b, ok := bf.(*blockSplitBloomFilter)
+ if !ok {
+ return
+ }
+
+ b.bitset32 = nil
+
+ if r.bufferPool != nil && b.cancelCleanup != nil {
+ // Stop the GC cleanup so it can't return b.data to the pool a second time.
+ b.cancelCleanup()
+ b.cancelCleanup = nil
+ r.bufferPool.Put(b.data)
Review Comment:
needs a call to `ResizeNoShrink(0)` first
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and benchmark [arrow-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #864:
URL: https://github.com/apache/arrow-go/pull/864
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org