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 531d2534 fix(parquet): explicitly release bloom builders (#1280)
531d2534 is described below
commit 531d2534893f651cb26d6901f70b197df0e170c0
Author: Matt Topol <[email protected]>
AuthorDate: Thu Sep 3 11:08:10 2026 -0400
fix(parquet): explicitly release bloom builders (#1280)
### Rationale for this change
Bloom-filter allocator tests rely on GC finalizers/cleanups and
immediately assert that all memory was released. GC does not wait for
cleanup callbacks, so `TestNewBloomFilter` intermittently reports
64-byte to 2 MiB leaks even though cleanup runs later.
### What changes are included in this PR?
- Add an idempotent `Release` method to `BloomFilterBuilder`
- Implement explicit release for block-split and adaptive builders
- Reuse the release path when adaptive candidates are discarded or
finalized
- Replace GC-dependent allocator assertions with deterministic release
The existing GC cleanup remains as a fallback for callers that do not
explicitly release a builder.
### Are these changes tested?
- `go test ./parquet/metadata -run
'Test(NewBloomFilter|AdaptiveBlockSplitBloomFilter)' -count=10`
- `go test ./parquet/...`
- pre-commit hooks
### Are there any user-facing changes?
`BloomFilterBuilder` now exposes `Release`, allowing callers to free
builder buffers deterministically.
Signed-off-by: Matt Topol <[email protected]>
---
parquet/metadata/adaptive_bloom_filter.go | 22 ++++++++++++----------
parquet/metadata/bloom_filter.go | 15 +++++++++++++++
parquet/metadata/bloom_filter_test.go | 17 +++++------------
3 files changed, 32 insertions(+), 22 deletions(-)
diff --git a/parquet/metadata/adaptive_bloom_filter.go
b/parquet/metadata/adaptive_bloom_filter.go
index 1966af06..9dcbedae 100644
--- a/parquet/metadata/adaptive_bloom_filter.go
+++ b/parquet/metadata/adaptive_bloom_filter.go
@@ -131,8 +131,7 @@ func (b *adaptiveBlockSplitBloomFilter) InsertHash(hash
uint64) {
b.candidates = slices.DeleteFunc(b.candidates, func(c
*bloomFilterCandidate) bool {
if c.expectedNDV < uint32(b.numDistinct) && c !=
b.largestCandidate {
- c.bloomFilter.cancelCleanup()
- c.bloomFilter.data.Release()
+ c.bloomFilter.Release()
return true
}
return false
@@ -161,8 +160,7 @@ func (b *adaptiveBlockSplitBloomFilter) InsertBulk(hashes
[]uint64) {
b.candidates = slices.DeleteFunc(b.candidates, func(c
*bloomFilterCandidate) bool {
if c.expectedNDV < uint32(b.numDistinct) && c !=
b.largestCandidate {
- c.bloomFilter.cancelCleanup()
- c.bloomFilter.data.Release()
+ c.bloomFilter.Release()
return true
}
return false
@@ -181,18 +179,22 @@ func (b *adaptiveBlockSplitBloomFilter) CheckHash(hash
uint64) bool {
return b.largestCandidate.bloomFilter.CheckHash(hash)
}
+func (b *adaptiveBlockSplitBloomFilter) Release() {
+ for _, c := range b.candidates {
+ c.bloomFilter.Release()
+ }
+ b.candidates = nil
+ b.largestCandidate = nil
+ b.finalized = true
+}
+
func (b *adaptiveBlockSplitBloomFilter) WriteTo(w io.Writer, enc
encryption.Encryptor) (int, error) {
b.finalized = true
optimal := b.optimalCandidate()
n, err := optimal.bloomFilter.WriteTo(w, enc)
- for _, c := range b.candidates {
- c.bloomFilter.cancelCleanup()
- c.bloomFilter.data.Release()
- }
- b.candidates = nil
- b.largestCandidate = nil
+ b.Release()
return n, err
}
diff --git a/parquet/metadata/bloom_filter.go b/parquet/metadata/bloom_filter.go
index 0dfbde9b..c4466617 100644
--- a/parquet/metadata/bloom_filter.go
+++ b/parquet/metadata/bloom_filter.go
@@ -581,6 +581,18 @@ func (b *blockSplitBloomFilter) Size() int64 {
return int64(len(b.bitset32) * 4)
}
+func (b *blockSplitBloomFilter) Release() {
+ if b.cancelCleanup != nil {
+ b.cancelCleanup()
+ b.cancelCleanup = nil
+ }
+ if b.data != nil {
+ b.data.Release()
+ b.data = nil
+ b.bitset32 = nil
+ }
+}
+
func (b *blockSplitBloomFilter) WriteTo(w io.Writer, enc encryption.Encryptor)
(int, error) {
if enc != nil {
n := enc.Encrypt(w, b.data.Bytes())
@@ -630,6 +642,9 @@ func NewBloomFilterFromNDVAndFPP(ndv uint32, fpp float64,
maxBytes int64, mem me
type BloomFilterBuilder interface {
Hasher() Hasher
Size() int64
+ // Release immediately frees buffers owned by the builder. It is safe to
+ // call more than once. Builders otherwise release their buffers during
GC.
+ Release()
InsertHash(hash uint64)
InsertBulk(hashes []uint64)
WriteTo(io.Writer, encryption.Encryptor) (int, error)
diff --git a/parquet/metadata/bloom_filter_test.go
b/parquet/metadata/bloom_filter_test.go
index 720de535..c2b210d0 100644
--- a/parquet/metadata/bloom_filter_test.go
+++ b/parquet/metadata/bloom_filter_test.go
@@ -196,12 +196,10 @@ func TestNewBloomFilter(t *testing.T) {
mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
defer mem.AssertSize(t, 0)
- {
- bf := NewBloomFilterFromNDVAndFPP(tt.ndv,
tt.fpp, tt.maxBytes, mem)
- assert.EqualValues(t, tt.expectedBytes,
bf.Size())
- runtime.GC()
- }
- runtime.GC() // force GC to run and do the cleanup
routines
+ bf := NewBloomFilterFromNDVAndFPP(tt.ndv, tt.fpp,
tt.maxBytes, mem)
+ assert.EqualValues(t, tt.expectedBytes, bf.Size())
+ bf.Release()
+ bf.Release() // releasing a builder is idempotent
})
}
}
@@ -351,12 +349,7 @@ func TestAdaptiveBloomFilterEdgeCases(t *testing.T) {
t.Run("clamps maximum size to the minimum allocation", func(t
*testing.T) {
bf := NewAdaptiveBlockSplitBloomFilter(0, 1, 0.01, col,
mem).(*adaptiveBlockSplitBloomFilter)
- defer func() {
- for _, candidate := range bf.candidates {
- candidate.bloomFilter.cancelCleanup()
- candidate.bloomFilter.data.Release()
- }
- }()
+ defer bf.Release()
assert.EqualValues(t, minimumBloomFilterBytes, bf.maxBytes)
assert.NotPanics(t, func() { bf.InsertHash(1) })
