This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new e126450c fix(parquet/metadata): fix BloomFilter memory recycling, doc
contract, and benchmark (#864)
e126450c is described below
commit e126450cba3a189de288fe679ed2f6df02d039e0
Author: Patzifist <[email protected]>
AuthorDate: Fri Jul 10 19:34:46 2026 +0300
fix(parquet/metadata): fix BloomFilter memory recycling, doc contract, and
benchmark (#864)
Adds a scoped VisitColumnBloomFilter accessor that recycles bloom-filter
buffers through a pool to avoid GC-driven memory spikes on very large Parquet
files, fixes sync.Pool pollution and unbounded heap growth in
GetColumnBloomFilter, resets recycled buffers before returning them to the
pool, and adds benchmark and regression coverage.
---
parquet/metadata/bloom_filter.go | 43 ++++++++
parquet/metadata/bloom_filter_benchmark_test.go | 140 ++++++++++++++++++++++++
parquet/metadata/bloom_filter_test.go | 51 +++++++++
3 files changed, 234 insertions(+)
diff --git a/parquet/metadata/bloom_filter.go b/parquet/metadata/bloom_filter.go
index 562d0d44..bd9faaec 100644
--- a/parquet/metadata/bloom_filter.go
+++ b/parquet/metadata/bloom_filter.go
@@ -531,6 +531,8 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i
int) (BloomFilter, er
}
if _, err = io.ReadFull(sectionRdr,
buf.Bytes()[filterBytesInHeader:]); err != nil {
+ buf.ResizeNoShrink(0)
+ r.bufferPool.Put(buf)
return nil, err
}
bitset = buf.Bytes()
@@ -552,6 +554,47 @@ 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
+ b.data.ResizeNoShrink(0)
+ r.bufferPool.Put(b.data)
+ } else if b.data != nil {
+ b.data.Release()
+ }
+
+ b.data = nil
+}
+
type FileBloomFilterBuilder struct {
Schema *schema.Schema
Encryptor encryption.FileEncryptor
diff --git a/parquet/metadata/bloom_filter_benchmark_test.go
b/parquet/metadata/bloom_filter_benchmark_test.go
new file mode 100644
index 00000000..15838b8a
--- /dev/null
+++ b/parquet/metadata/bloom_filter_benchmark_test.go
@@ -0,0 +1,140 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+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()
+
+ 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)
+
+ var payloadBuf bytes.Buffer
+ payloadBuf.Write(make([]byte, 8))
+
+ rnd := rand.New(rand.NewSource(42))
+ for i := 0; i < metaCount; i++ {
+ bloomFilterDataSize := int32(1*1024*1024 + rnd.Intn(200*1024))
+
+ header := format.BloomFilterHeader{
+ NumBytes: bloomFilterDataSize,
+ Algorithm: &defaultAlgorithm,
+ Hash: &defaultHashStrategy,
+ Compression: &defaultCompression,
+ }
+
+ serializer := thrift.NewThriftSerializer()
+ headerBytes, _ := serializer.Write(ctx, &header)
+
+ currentOffset := int64(payloadBuf.Len())
+ payloadBuf.Write(headerBytes)
+
+ payloadBuf.Write(make([]byte, bloomFilterDataSize))
+
+ bloomFilterReadSize := int32(len(headerBytes))
+
+ columnMetaData := format.ColumnMetaData{
+ Type: format.Type_BYTE_ARRAY,
+ PathInSchema: []string{"test_col"},
+ Codec: format.CompressionCodec_UNCOMPRESSED,
+ BloomFilterOffset: ¤tOffset,
+ BloomFilterLength: &bloomFilterReadSize,
+ }
+ thriftColumnChunk := format.ColumnChunk{MetaData:
&columnMetaData}
+ thriftRowGroup := format.RowGroup{Columns:
[]*format.ColumnChunk{&thriftColumnChunk}}
+ metaList[i] = NewRowGroupMetaData(&thriftRowGroup, sc, nil, nil)
+ }
+
+ payload := payloadBuf.Bytes()
+ 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)
+ }
+ }
+ })
+}
diff --git a/parquet/metadata/bloom_filter_test.go
b/parquet/metadata/bloom_filter_test.go
index 011f67a5..86126502 100644
--- a/parquet/metadata/bloom_filter_test.go
+++ b/parquet/metadata/bloom_filter_test.go
@@ -20,8 +20,10 @@ import (
"fmt"
"math/rand/v2"
"runtime"
+ "sync"
"testing"
+ "github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/bitutil"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
@@ -721,3 +723,52 @@ 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{}
+ 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()
+ if got != nil {
+ t.Fatalf("encrypted buffer should not be recycled into the
pool, but got: %v", got)
+ }
+}
+
+func TestRecycleUnencryptedPathPool(t *testing.T) {
+ pool := &sync.Pool{
+ New: func() any {
+ return
memory.NewResizableBuffer(memory.NewGoAllocator())
+ },
+ }
+ r := &RowGroupBloomFilterReader{bufferPool: pool}
+
+ buf := pool.Get().(*memory.Buffer)
+ buf.ResizeNoShrink(1024)
+
+ bf := &blockSplitBloomFilter{
+ data: buf,
+ bitset32: arrow.Uint32Traits.CastFromBytes(buf.Bytes()),
+ }
+
+ addCleanup(bf, r.bufferPool)
+
+ r.recycle(bf)
+
+ runtime.GC()
+ runtime.GC()
+
+ got1 := pool.Get().(*memory.Buffer)
+ got2 := pool.Get().(*memory.Buffer)
+
+ if got1 == got2 {
+ t.Fatalf("Pool pollution detected! The same buffer instance was
put into the pool twice.")
+ }
+}
