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 6d0e70cf fix(parquet/metadata): validate adaptive bloom options (#1033)
6d0e70cf is described below
commit 6d0e70cf73d2c403ba2a23e6778681e854f7abd9
Author: Minh Vu <[email protected]>
AuthorDate: Tue Jul 28 17:07:17 2026 +0200
fix(parquet/metadata): validate adaptive bloom options (#1033)
### Rationale for this change
The adaptive bloom filter constructor accepted invalid configuration
values. A false-positive probability of one could loop indefinitely in
non-assertion builds, while a maximum size below the minimum could
create a zero-byte filter that later panicked during insertion.
### What changes are included in this PR?
* Reject false-positive probabilities outside the open interval `(0,
1)`, including NaN.
* Clamp the maximum allocation to the supported bloom-filter size range.
* Keep the existing constructor signature and small-size fallback
behavior.
### Are these changes tested?
Yes. The tests cover invalid probabilities and a zero-byte requested
maximum, including insertion into the resulting minimum-sized filter.
The full `parquet/metadata` package passes.
---
parquet/metadata/adaptive_bloom_filter.go | 7 +++++-
parquet/metadata/bloom_filter.go | 26 +++++----------------
parquet/metadata/bloom_filter_test.go | 38 ++++++++++++++++++++++++++++++-
parquet/reader_writer_properties_test.go | 22 ++++++++++++++++++
parquet/writer_properties.go | 16 +++++++++++++
5 files changed, 87 insertions(+), 22 deletions(-)
diff --git a/parquet/metadata/adaptive_bloom_filter.go
b/parquet/metadata/adaptive_bloom_filter.go
index 39fee2d7..1966af06 100644
--- a/parquet/metadata/adaptive_bloom_filter.go
+++ b/parquet/metadata/adaptive_bloom_filter.go
@@ -18,6 +18,7 @@ package metadata
import (
"io"
+ "math"
"slices"
"github.com/apache/arrow-go/v18/arrow"
@@ -79,9 +80,13 @@ type adaptiveBlockSplitBloomFilter struct {
}
func NewAdaptiveBlockSplitBloomFilter(maxBytes uint32, numCandidates int, fpp
float64, column *schema.Column, mem memory.Allocator) BloomFilterBuilder {
+ if fpp <= 0 || fpp >= 1 || math.IsNaN(fpp) {
+ panic("parquet: bloom filter false-positive probability must be
in (0, 1)")
+ }
+ maxBytes = max(minimumBloomFilterBytes, min(maximumBloomFilterBytes,
maxBytes))
ret := &adaptiveBlockSplitBloomFilter{
mem: mem,
- maxBytes: min(maximumBloomFilterBytes, maxBytes),
+ maxBytes: maxBytes,
minBytes: minimumBloomFilterBytes,
minCandidateNDV: 16,
hasher: xxhasher{},
diff --git a/parquet/metadata/bloom_filter.go b/parquet/metadata/bloom_filter.go
index cd2c79ff..100786e1 100644
--- a/parquet/metadata/bloom_filter.go
+++ b/parquet/metadata/bloom_filter.go
@@ -293,14 +293,11 @@ func (b *blockSplitBloomFilter) WriteTo(w io.Writer, enc
encryption.Encryptor) (
}
func NewBloomFilter(numBytes, maxBytes uint32, mem memory.Allocator)
BloomFilterBuilder {
+ maxBytes = max(minimumBloomFilterBytes, min(maximumBloomFilterBytes,
maxBytes))
if numBytes < minimumBloomFilterBytes {
numBytes = minimumBloomFilterBytes
}
- if maxBytes > maximumBloomFilterBytes {
- maxBytes = maximumBloomFilterBytes
- }
-
if numBytes > maxBytes {
numBytes = maxBytes
}
@@ -325,23 +322,12 @@ func NewBloomFilter(numBytes, maxBytes uint32, mem
memory.Allocator) BloomFilter
}
func NewBloomFilterFromNDVAndFPP(ndv uint32, fpp float64, maxBytes int64, mem
memory.Allocator) BloomFilterBuilder {
- numBytes := optimalNumBytes(ndv, fpp)
- if numBytes > uint32(maxBytes) {
- numBytes = uint32(maxBytes)
+ if fpp <= 0 || fpp >= 1 || math.IsNaN(fpp) {
+ panic("parquet: bloom filter false-positive probability must be
in (0, 1)")
}
-
- buf := memory.NewResizableBuffer(mem)
- buf.ResizeNoShrink(int(numBytes))
- bf := &blockSplitBloomFilter{
- data: buf,
- bitset32: arrow.Uint32Traits.CastFromBytes(buf.Bytes()),
- hasher: xxhasher{},
- algorithm: format.BloomFilterAlgorithm{BLOCK:
&format.SplitBlockAlgorithm{}},
- hashStrategy: format.BloomFilterHash{XXHASH: &format.XxHash{}},
- compression: format.BloomFilterCompression{UNCOMPRESSED:
&format.Uncompressed{}},
- }
- addCleanup(bf, nil)
- return bf
+ maxBytes = max(minimumBloomFilterBytes, min(maximumBloomFilterBytes,
maxBytes))
+ numBytes := optimalNumBytes(ndv, fpp)
+ return NewBloomFilter(numBytes, uint32(maxBytes), mem)
}
type BloomFilterBuilder interface {
diff --git a/parquet/metadata/bloom_filter_test.go
b/parquet/metadata/bloom_filter_test.go
index 955b7324..4206ec56 100644
--- a/parquet/metadata/bloom_filter_test.go
+++ b/parquet/metadata/bloom_filter_test.go
@@ -18,6 +18,7 @@ package metadata
import (
"fmt"
+ "math"
"math/rand/v2"
"runtime"
"sync"
@@ -120,7 +121,7 @@ func TestNewBloomFilter(t *testing.T) {
maxBytes int64
expectedBytes int64
}{
- {1, 0.09, 0, 0},
+ {1, 0.09, 0, minimumBloomFilterBytes},
// cap at maximumBloomFilterBytes
{1 << 30, 0.9, maximumBloomFilterBytes + 1,
maximumBloomFilterBytes},
// round to power of 2
@@ -144,6 +145,20 @@ func TestNewBloomFilter(t *testing.T) {
}
}
+func TestNewBloomFilterFromNDVAndFPPValidatesOptions(t *testing.T) {
+ for _, fpp := range []float64{0, 1, -0.1, 2, math.NaN()} {
+ t.Run(fmt.Sprintf("fpp=%v", fpp), func(t *testing.T) {
+ assert.Panics(t, func() {
+ NewBloomFilterFromNDVAndFPP(1, fpp, 1024,
memory.DefaultAllocator)
+ })
+ })
+ }
+
+ bf := NewBloomFilterFromNDVAndFPP(1, 0.01, 0, memory.DefaultAllocator)
+ assert.EqualValues(t, minimumBloomFilterBytes, bf.Size())
+ assert.NotPanics(t, func() { bf.InsertHash(42) })
+}
+
func BenchmarkFilterInsert(b *testing.B) {
bf := blockSplitBloomFilter{bitset32: make([]uint32, 8)}
b.ResetTimer()
@@ -272,6 +287,27 @@ func TestAdaptiveBloomFilterEdgeCases(t *testing.T) {
assert.Truef(t, bf.CheckHash(h), "hash %d not found
after GC - potential GC safety issue", h)
}
})
+
+ 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()
+ }
+ }()
+
+ assert.EqualValues(t, minimumBloomFilterBytes, bf.maxBytes)
+ assert.NotPanics(t, func() { bf.InsertHash(1) })
+ })
+
+ t.Run("rejects invalid false-positive probabilities", func(t
*testing.T) {
+ for _, fpp := range []float64{-0.1, 0, 1, math.NaN()} {
+ assert.PanicsWithValue(t,
+ "parquet: bloom filter false-positive
probability must be in (0, 1)",
+ func() { NewAdaptiveBlockSplitBloomFilter(1024,
1, fpp, col, mem) })
+ }
+ })
}
func TestAdaptiveBloomFilterEndToEnd(t *testing.T) {
diff --git a/parquet/reader_writer_properties_test.go
b/parquet/reader_writer_properties_test.go
index 00b26a83..ee8ac78f 100644
--- a/parquet/reader_writer_properties_test.go
+++ b/parquet/reader_writer_properties_test.go
@@ -18,6 +18,7 @@ package parquet_test
import (
"bytes"
+ "math"
"testing"
"github.com/apache/arrow-go/v18/arrow/memory"
@@ -62,6 +63,27 @@ func TestWriterPropAdvanced(t *testing.T) {
assert.Equal(t, parquet.Repetitions.Required, props.RootRepetition())
}
+func TestBloomFilterPropertiesValidateOptions(t *testing.T) {
+ for _, fpp := range []float64{0, 1, -0.1, 2, math.NaN()} {
+ t.Run("invalid false-positive probability", func(t *testing.T) {
+ assert.Panics(t, func() {
+
parquet.NewWriterProperties(parquet.WithBloomFilterFPP(fpp))
+ })
+ assert.Panics(t, func() {
+
parquet.NewWriterProperties(parquet.WithBloomFilterFPPFor("column", fpp))
+ })
+ })
+ }
+
+ for _, maxBytes := range []int64{0, 31, 128*1024*1024 + 1} {
+ t.Run("invalid maximum size", func(t *testing.T) {
+ assert.Panics(t, func() {
+
parquet.NewWriterProperties(parquet.WithMaxBloomFilterBytes(maxBytes))
+ })
+ })
+ }
+}
+
func TestReaderPropsGetStreamInsufficient(t *testing.T) {
data := "shorter than expected"
buf := memory.NewBufferBytes([]byte(data))
diff --git a/parquet/writer_properties.go b/parquet/writer_properties.go
index 4f036014..4327bcf1 100644
--- a/parquet/writer_properties.go
+++ b/parquet/writer_properties.go
@@ -17,6 +17,8 @@
package parquet
import (
+ "math"
+
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet/compress"
@@ -58,6 +60,9 @@ const (
DefaultBloomFilterFPP = 0.01
DefaultAdaptiveBloomFilterEnabled = false
DefaultBloomFilterCandidates = 5
+
+ minimumBloomFilterBytes = 32
+ maximumBloomFilterBytes = 128 * 1024 * 1024
)
// ColumnProperties defines the encoding, codec, and so on for a given column.
@@ -366,6 +371,9 @@ func WithPageIndexEnabledPath(path ColumnPath, enabled
bool) WriterProperty {
// it is abandoned and not written to the file.
func WithMaxBloomFilterBytes(nbytes int64) WriterProperty {
return func(cfg *writerPropConfig) {
+ if nbytes < minimumBloomFilterBytes || nbytes >
maximumBloomFilterBytes {
+ panic("parquet: maximum bloom filter size must be
between 32 bytes and 128 MiB")
+ }
cfg.wr.maxBloomFilterBytes = nbytes
}
}
@@ -396,6 +404,7 @@ func WithBloomFilterEnabledPath(path ColumnPath, enabled
bool) WriterProperty {
// bloom filters.
func WithBloomFilterFPP(fpp float64) WriterProperty {
return func(cfg *writerPropConfig) {
+ validateBloomFilterFPP(fpp)
cfg.wr.defColumnProps.BloomFilterFPP = fpp
}
}
@@ -404,10 +413,17 @@ func WithBloomFilterFPP(fpp float64) WriterProperty {
// for writing bloom filters.
func WithBloomFilterFPPFor(path string, fpp float64) WriterProperty {
return func(cfg *writerPropConfig) {
+ validateBloomFilterFPP(fpp)
cfg.bloomFilterFPPs[path] = fpp
}
}
+func validateBloomFilterFPP(fpp float64) {
+ if fpp <= 0 || fpp >= 1 || math.IsNaN(fpp) {
+ panic("parquet: bloom filter false-positive probability must be
in (0, 1)")
+ }
+}
+
// WithBloomFilterFPPPath is like WithBloomFilterFPPFor, but takes a ColumnPath
func WithBloomFilterFPPPath(path ColumnPath, fpp float64) WriterProperty {
return WithBloomFilterFPPFor(path.String(), fpp)
