Repository: segmentio/parquet-go
Pull request #53 (2022-01-28 08:39:40)

[O11Y-196] implement parquet bloom filters
This PR adds support for generating and querying parquet bloom filters as defined by the parquet format specification in https://github.com/apache/parquet-format/blob/master/BloomFilter.md

The flavor of bloom filters implemented in parquet are "split block bloom filters" (SBBF). The main difference with classic bloom filters is in the way the filter is divided in fixed-size blocks: when a key is inserted in the filter, its filter bits are written to a single 32 bytes block rather than being dispatched across the whole filter. This design guarantees that the number of CPU cache miss remains constant (at most one) no matter how many bits per key the filter is configured to host.

## Design

I designed the APIs to abstract away most of the complexity of generating and using the filters. To create filters, the program adds a `parquet.BloomFilters` option to the `parquet.Writer` configuration, enumerating the type filters to apply to each column, for example:

```go
writer := parquet.NewWriter(output,
  parquet.BloomFilters(
    parquet.SplitBlockFilter("last_name"),
  ),
)
```

I opted for a default behavior where no bloom filters are generated, because they can quickly increase the CPU and memory footprint of the program, and applications should generally know which columns they want to create filters for.

When reading parquet files, filters can be used to determine whether a column chunk may contain a value, by accessing the filter via the `parquet.ColumnChunk.BloomFilter` method, the general pattern looks as such:

```go
lastName := rowGroup.Column(lastNameColumnIndex)
bloomFilter := lastName.BloomFilter()

if ok, err := bloomFilter.Check([]byte("Solo")); err != nil {
  ...
} else if ok {
  // the value likely exists in the column chunk
  ...
}
```

The code for the bloom filter implementation is located in the new `bloom` package.

## Implementation

In past experiences we had using bloom filters in combination with LSM trees, we observed that filter generation accounted for over half of the compute time when performing compaction of runs.

I invested in research time to figure out how to design bloom filter APIs to reduce the overhead of filter generation (inserting keys), which led me down a rabbit hole of trying to determine how much throughput was possible to achieve to have a target baseline.

It turned out that the parquet bloom filter algorithm were well designed to leverage vectorized CPU instructions, so I added both pure Go and assembly versions of those functions, the vectorized versions of the algorithms yield large improvements to the throughput of the filter compared to the code generated by the Go compiler. Here is an example taken from one of the benchmarks I added where we insert 1000 64 bits values per iteration:

```
name                old time/op   new time/op    delta
BloomFilterEncoder   13.1µs ± 3%     2.7µs ± 5%   -79.19%  (p=0.000 n=10+10)

name                old speed     new speed      delta
BloomFilterEncoder  609MB/s ± 3%  2929MB/s ± 5%  +380.89%  (p=0.000 n=10+10)
```

Before I go further, it is important to mention that parquet bloom filter are somewhat composed of two steps: to insert a key, the XXH64 hash of the value is first computed, then the resulting 64 bits key is inserted in the filter.

I first started the implementation with https://github.com/cespare/xxhash but after optimizing the bloom filter to support bulk inserts (which created more opportunities to leverage vectorization), computing the hash of individual keys accounted for most of the CPU time spent. I attempted to optimize the hash function without success, so instead I pulled the `xxhash` package into parquet-go and made modifications to introduce new APIs that support bulk computation of hashes for fixed size keys (most use cases for bloom filters would likely be on columns holding primary keys that are often integers or UUIDs). Having those specialized APIs and working on bulks of inputs made it possible to leverage vectorization to increase the throughput of XXH64 computation, which yielded a second round of optimization.

This made it clear that the higher level APIs that rely on these primitives would also need to support writing values to the filters in bulks in order to benefit from the improvements. I managed to reuse the `encoding.Encoder` interface for this purpose, which allows pages of parquet values to be written in bulk to the filter in the same way that they are written to the file.

I still need to work on documenting how to configure and use the bloom filter APIs, but overall the code should be ready for feedback!

---
Edit: I ended up removing the vectorized XXH64 code; I ran more benchmarks and the end results yield mixed returns, which doesn't seem worth the added complexity. Assembly code is hard-enough to maintain, it really needs to be worth the cost; here is the comparison to the scalar vs vectorized assembly:

```
name                    old speed      new speed      delta
MultiSum64Uint8/16B     5.55GB/s ± 1%  5.84GB/s ± 2%   +5.13%  (p=0.000 n=8+9)
MultiSum64Uint8/100B    5.76GB/s ± 1%  6.46GB/s ± 5%  +12.04%  (p=0.000 n=9+9)
MultiSum64Uint8/4KB     6.14GB/s ± 2%  6.63GB/s ± 3%   +7.98%  (p=0.000 n=9+10)
MultiSum64Uint8/10MB    5.85GB/s ± 3%  6.34GB/s ± 3%   +8.43%  (p=0.000 n=10+10)
MultiSum64Uint16/16B    3.67GB/s ± 4%  4.16GB/s ± 2%  +13.37%  (p=0.000 n=10+10)
MultiSum64Uint16/100B   4.16GB/s ± 3%  4.47GB/s ± 3%   +7.41%  (p=0.000 n=10+10)
MultiSum64Uint16/4KB    4.39GB/s ± 1%  4.61GB/s ± 2%   +4.85%  (p=0.000 n=9+9)
MultiSum64Uint16/10MB   4.26GB/s ± 4%  4.39GB/s ± 3%   +2.88%  (p=0.009 n=10+10)
MultiSum64Uint32/4B     1.94GB/s ± 5%  1.69GB/s ± 2%  -12.89%  (p=0.000 n=10+9)
MultiSum64Uint32/16B    3.80GB/s ± 3%  3.58GB/s ± 6%   -5.70%  (p=0.000 n=10+10)
MultiSum64Uint32/100B   5.25GB/s ± 3%  5.78GB/s ± 3%  +10.10%  (p=0.000 n=10+10)
MultiSum64Uint32/4KB    5.59GB/s ± 2%  6.34GB/s ± 3%  +13.25%  (p=0.000 n=10+10)
MultiSum64Uint32/10MB   5.17GB/s ± 5%  5.83GB/s ± 5%  +12.95%  (p=0.000 n=10+10)
MultiSum64Uint64/16B    2.59GB/s ± 4%  2.28GB/s ± 4%  -11.72%  (p=0.000 n=10+10)
MultiSum64Uint64/100B   3.97GB/s ± 6%  4.03GB/s ± 3%     ~     (p=0.631 n=10+10)
MultiSum64Uint64/4KB    4.39GB/s ± 3%  4.97GB/s ± 5%  +13.23%  (p=0.000 n=10+10)
MultiSum64Uint64/10MB   4.09GB/s ± 1%  4.47GB/s ± 5%   +9.15%  (p=0.000 n=8+10)
MultiSum64Uint128/16B   1.56GB/s ± 2%  1.52GB/s ± 5%   -2.92%  (p=0.021 n=8+10)
MultiSum64Uint128/100B  2.57GB/s ± 4%  2.66GB/s ± 5%   +3.64%  (p=0.010 n=9+10)
MultiSum64Uint128/4KB   3.13GB/s ± 3%  3.06GB/s ± 9%     ~     (p=0.579 n=10+10)
MultiSum64Uint128/10MB  2.84GB/s ± 3%  2.79GB/s ± 1%   -1.90%  (p=0.003 n=9+9)

name                    old hash/s     new hash/s     delta
MultiSum64Uint8/16B         694M ± 1%      730M ± 2%   +5.13%  (p=0.000 n=8+9)
MultiSum64Uint8/100B        720M ± 1%      807M ± 5%  +12.04%  (p=0.000 n=9+9)
MultiSum64Uint8/4KB         768M ± 2%      829M ± 3%   +7.98%  (p=0.000 n=9+10)
MultiSum64Uint8/10MB        731M ± 3%      792M ± 3%   +8.43%  (p=0.000 n=10+10)
MultiSum64Uint16/16B        458M ± 4%      519M ± 2%  +13.37%  (p=0.000 n=10+10)
MultiSum64Uint16/100B       520M ± 3%      559M ± 3%   +7.41%  (p=0.000 n=10+10)
MultiSum64Uint16/4KB        549M ± 1%      576M ± 2%   +4.85%  (p=0.000 n=9+9)
MultiSum64Uint16/10MB       533M ± 4%      548M ± 3%   +2.88%  (p=0.009 n=10+10)
MultiSum64Uint32/4B         243M ± 5%      211M ± 2%  -12.89%  (p=0.000 n=10+9)
MultiSum64Uint32/16B        475M ± 3%      448M ± 6%   -5.70%  (p=0.000 n=10+10)
MultiSum64Uint32/100B       657M ± 3%      723M ± 3%  +10.10%  (p=0.000 n=10+10)
MultiSum64Uint32/4KB        699M ± 2%      792M ± 3%  +13.25%  (p=0.000 n=10+10)
MultiSum64Uint32/10MB       646M ± 5%      729M ± 5%  +12.95%  (p=0.000 n=10+10)
MultiSum64Uint64/16B        323M ± 4%      285M ± 4%  -11.72%  (p=0.000 n=10+10)
MultiSum64Uint64/100B       496M ± 6%      503M ± 3%     ~     (p=0.631 n=10+10)
MultiSum64Uint64/4KB        548M ± 3%      621M ± 5%  +13.23%  (p=0.000 n=10+10)
MultiSum64Uint64/10MB       512M ± 1%      559M ± 5%   +9.15%  (p=0.000 n=8+10)
MultiSum64Uint128/16B       196M ± 2%      190M ± 5%   -2.92%  (p=0.021 n=8+10)
MultiSum64Uint128/100B      321M ± 4%      332M ± 5%   +3.64%  (p=0.010 n=9+10)
MultiSum64Uint128/4KB       391M ± 3%      382M ± 9%     ~     (p=0.579 n=10+10)
MultiSum64Uint128/10MB      355M ± 3%      349M ± 1%   -1.90%  (p=0.003 n=9+9)
```

I kept the code on a separate branch for now https://github.com/segmentio/parquet-go/tree/xxhash-avx2, I'll experiment with AVX-512 and ARM64 vectorization as follow up.

achille-roussel: Thanks for the thorough review @kevinburkesegment !

Patch:
diff --git a/README.md b/README.md
index fc6f494..0a89f0b 100644
--- a/README.md
+++ b/README.md
@@ -245,7 +245,7 @@ if err := writer.Close(); err != nil {
 }
 ```
 
-### Merging Row Groups: [parquet.MergeRowGroups](http://pkg.go.dev/github.com/segmentio/parquet-go#MergeRowGroups)
+### Merging Row Groups: [parquet.MergeRowGroups](https://pkg.go.dev/github.com/segmentio/parquet-go#MergeRowGroups)
 
 Parquet files are often used as part of the underlying engine for data
 processing or storage layers, in which cases merging multiple row groups
@@ -287,6 +287,72 @@ if err := writer.Close(); err != nil {
 }
 ```
 
+### Using Bloom Filters: [parquet.BloomFilter](https://pkg.go.dev/github.com/segmentio/parquet-go#BloomFilter)
+
+Parquet files can embed bloom filters to help improve the performance of point
+lookups in the files. The format of parquet bloom filters is documented in
+the parquet specification: [Parquet Bloom Filter](https://github.com/apache/parquet-format/blob/master/BloomFilter.md)
+
+By default, no bloom filters are created in parquet files, but applications can
+configure the list of columns to create filters for using the `parquet.BloomFilters`
+option when instantiating writers; for example:
+
+```go
+writer := parquet.NewWriter(output,
+    parquet.BloomFilters(
+        // Configures the write to generate split-block bloom filters for the
+        // "first_name" and "last_name" columns of the parquet schema of rows
+        // witten by the application.
+        parquet.SplitBlockFilter("first_name"),
+        parquet.SplitBlockFilter("last_name"),
+    ),
+)
+...
+```
+
+Generating bloom filters requires to know how many values exist in a column
+chunk in order to properly size the filter, which requires buffering all the
+values written to the column in memory. Because of it, the memory footprint
+of `parquet.Writer` increases linearly with the number of columns that the
+writer needs to generate filters for. This extra cost is optimized away when
+rows are copied from a `parquet.Buffer` to a writer, since in this case the
+number of values per column in known since the buffer already holds all the
+values in memory.
+
+When reading parquet files, column chunks expose the generated bloom filters
+with the `parquet.ColumnChunk.BloomFilter` method, returning a
+`parquet.BloomFilter` instance if a filter was available, or `nil` when there
+were no filters.
+
+Using bloom filters in parquet files is useful when performing point-lookups in
+parquet files; searching for column rows matching a given value. Programs can
+quickly eliminate column chunks that they know does not contain the value they
+search for by checking the filter first, which is often multiple orders of
+magnitude faster than scanning the column.
+
+The following code snippet hilights how filters are typically used:
+
+```go
+var candidateChunks []parquet.ColumnChunk
+
+for i, n := 0, file.NumRowGroups(); i < n; i++ {
+    columnChunk := file.RowGroup(i).Column(columnIndex)
+    bloomFilter := columnChunk.BloomFilter()
+
+    if bloomFilter != nil {
+        if ok, err := bloomFilter.Check(value); err != nil {
+            ...
+        } else if !ok {
+            // Bloom filters may return false positives, but never return false
+            // negatives, we know this column chunk does not contain the value.
+            continue
+        }
+    }
+
+    candidateChunks = append(candidateChunks, columnChunk)
+}
+```
+
 ## Maintenance
 
 The project is hosted and maintained by Twilio; we welcome external contributors
diff --git a/bloom.go b/bloom.go
new file mode 100644
index 0000000..17754db
--- /dev/null
+++ b/bloom.go
@@ -0,0 +1,243 @@
+package parquet
+
+import (
+	"io"
+	"unsafe"
+
+	"github.com/segmentio/parquet-go/bloom"
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/format"
+)
+
+// BloomFilter is an interface allowing applications to test whether a key
+// exists in a bloom filter.
+type BloomFilter interface {
+	// Implement the io.ReaderAt interface as a mechanism to allow reading the
+	// raw bits of the filter.
+	io.ReaderAt
+
+	// Returns the size of the bloom filter (in bytes).
+	Size() int64
+
+	// Tests whether the given value is present in the filter.
+	//
+	// The parquet value is expected to be PLAIN-encoded, as it would be by
+	// calling the parquet.Value.Bytes method.
+	//
+	// A non-nil error may be returned if reading the filter failed. This may
+	// happen if the filter was lazily loaded from a storage medium during the
+	// call to Check for example. Applications that can guarantee that the
+	// filter was in memory at the time Check was called can safely ignore the
+	// error, which would always be nil in this case.
+	Check(value []byte) (bool, error)
+}
+
+type bloomFilter struct {
+	io.SectionReader
+	hash  bloom.Hash
+	check func(io.ReaderAt, int64, uint64) (bool, error)
+}
+
+func (f *bloomFilter) Check(value []byte) (bool, error) {
+	return f.check(&f.SectionReader, f.Size(), f.hash.Sum64(value))
+}
+
+func newBloomFilter(file io.ReaderAt, offset int64, header *format.BloomFilterHeader) *bloomFilter {
+	if header.Algorithm.Block != nil {
+		if header.Hash.XxHash != nil {
+			if header.Compression.Uncompressed != nil {
+				return &bloomFilter{
+					SectionReader: *io.NewSectionReader(file, offset, int64(header.NumBytes)),
+					hash:          bloom.XXH64{},
+					check:         bloom.CheckSplitBlock,
+				}
+			}
+		}
+	}
+	return nil
+}
+
+// The BloomFilterColumn interface is a declarative representation of bloom filters
+// used when configuring filters on a parquet writer.
+type BloomFilterColumn interface {
+	// Returns the path of the column that the filter applies to.
+	Path() []string
+
+	// Returns the hashing algorithm used when inserting values into a bloom
+	// filter.
+	Hash() bloom.Hash
+
+	// NewFilter constructs a new bloom filter configured to hold the given
+	// number of values and bits of filter per value.
+	NewFilter(numValues int64, bitsPerValue uint) bloom.MutableFilter
+}
+
+// SplitBlockFilter constructs a split block bloom filter object for the column
+// at the given path.
+func SplitBlockFilter(path ...string) BloomFilterColumn { return splitBlockFilter(path) }
+
+type splitBlockFilter []string
+
+func (f splitBlockFilter) Path() []string   { return f }
+func (f splitBlockFilter) Hash() bloom.Hash { return bloom.XXH64{} }
+func (f splitBlockFilter) NewFilter(numValues int64, bitsPerValue uint) bloom.MutableFilter {
+	return make(bloom.SplitBlockFilter, bloom.NumSplitBlocksOf(numValues, bitsPerValue))
+}
+
+// Creates a header from the given bloom filter.
+//
+// For now there is only one type of filter supported, but we provide this
+// function to suggest a model for extending the implementation if new filters
+// are added to the parquet specs.
+func bloomFilterHeader(filter BloomFilterColumn) (header format.BloomFilterHeader) {
+	switch filter.(type) {
+	case splitBlockFilter:
+		header.Algorithm.Block = &format.SplitBlockAlgorithm{}
+	}
+	switch filter.Hash().(type) {
+	case bloom.XXH64:
+		header.Hash.XxHash = &format.XxHash{}
+	}
+	header.Compression.Uncompressed = &format.BloomFilterUncompressed{}
+	return header
+}
+
+func searchBloomFilterColumn(filters []BloomFilterColumn, path columnPath) BloomFilterColumn {
+	for _, f := range filters {
+		if path.equal(f.Path()) {
+			return f
+		}
+	}
+	return nil
+}
+
+// bloomFilterEncoder is an adapter type which implements the encoding.Encoder
+// interface on top of a bloom filter.
+type bloomFilterEncoder struct {
+	filter bloom.MutableFilter
+	hash   bloom.Hash
+	keys   [128]uint64
+}
+
+func newBloomFilterEncoder(filter bloom.MutableFilter, hash bloom.Hash) *bloomFilterEncoder {
+	return &bloomFilterEncoder{filter: filter, hash: hash}
+}
+
+func (e *bloomFilterEncoder) Check(value []byte) bool {
+	return e.filter.Check(e.hash.Sum64(value))
+}
+
+func (e *bloomFilterEncoder) Bytes() []byte {
+	return e.filter.Bytes()
+}
+
+func (e *bloomFilterEncoder) Reset(io.Writer) {
+	e.filter.Reset()
+}
+
+func (e *bloomFilterEncoder) SetBitWidth(int) {
+}
+
+func (e *bloomFilterEncoder) EncodeBoolean(data []bool) error {
+	return e.insert8(*(*[]uint8)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeInt8(data []int8) error {
+	return e.insert8(*(*[]uint8)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeInt16(data []int16) error {
+	return e.insert16(*(*[]uint16)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeInt32(data []int32) error {
+	return e.insert32(*(*[]uint32)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeInt64(data []int64) error {
+	return e.insert64(*(*[]uint64)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeInt96(data []deprecated.Int96) error {
+	return e.EncodeFixedLenByteArray(12, deprecated.Int96ToBytes(data))
+}
+
+func (e *bloomFilterEncoder) EncodeFloat(data []float32) error {
+	return e.insert32(*(*[]uint32)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeDouble(data []float64) error {
+	return e.insert64(*(*[]uint64)(unsafe.Pointer(&data)))
+}
+
+func (e *bloomFilterEncoder) EncodeByteArray(data encoding.ByteArrayList) error {
+	data.Range(func(v []byte) bool { e.insert(v); return true })
+	return nil
+}
+
+func (e *bloomFilterEncoder) EncodeFixedLenByteArray(size int, data []byte) error {
+	if size == 16 {
+		return e.insert128(*(*[][16]byte)(unsafe.Pointer(&data)))
+	}
+	for i, j := 0, size; j <= len(data); {
+		e.insert(data[i:j])
+		i += size
+		j += size
+	}
+	return nil
+}
+
+func (e *bloomFilterEncoder) insert(value []byte) {
+	e.filter.Insert(e.hash.Sum64(value))
+}
+
+func (e *bloomFilterEncoder) insert8(data []uint8) error {
+	k := e.keys[:]
+	for i := 0; i < len(data); {
+		n := e.hash.MultiSum64Uint8(k, data[i:])
+		e.filter.InsertBulk(k[:n:n])
+		i += n
+	}
+	return nil
+}
+
+func (e *bloomFilterEncoder) insert16(data []uint16) error {
+	k := e.keys[:]
+	for i := 0; i < len(data); {
+		n := e.hash.MultiSum64Uint16(k, data[i:])
+		e.filter.InsertBulk(k[:n:n])
+		i += n
+	}
+	return nil
+}
+
+func (e *bloomFilterEncoder) insert32(data []uint32) error {
+	k := e.keys[:]
+	for i := 0; i < len(data); {
+		n := e.hash.MultiSum64Uint32(k, data[i:])
+		e.filter.InsertBulk(k[:n:n])
+		i += n
+	}
+	return nil
+}
+
+func (e *bloomFilterEncoder) insert64(data []uint64) error {
+	k := e.keys[:]
+	for i := 0; i < len(data); {
+		n := e.hash.MultiSum64Uint64(k, data[i:])
+		e.filter.InsertBulk(k[:n:n])
+		i += n
+	}
+	return nil
+}
+
+func (e *bloomFilterEncoder) insert128(data [][16]byte) error {
+	k := e.keys[:]
+	for i := 0; i < len(data); {
+		n := e.hash.MultiSum64Uint128(k, data[i:])
+		e.filter.InsertBulk(k[:n:n])
+		i += n
+	}
+	return nil
+}
diff --git a/bloom/block.go b/bloom/block.go
new file mode 100644
index 0000000..193dec7
--- /dev/null
+++ b/bloom/block.go
@@ -0,0 +1,28 @@
+package bloom
+
+import "unsafe"
+
+// Word represents 32 bits words of bloom filter blocks.
+type Word uint32
+
+// Block represents bloom filter blocks which contain eight 32 bits words.
+type Block [8]Word
+
+// Bytes returns b as a byte slice.
+func (b *Block) Bytes() []byte {
+	return unsafe.Slice((*byte)(unsafe.Pointer(b)), BlockSize)
+}
+
+const (
+	// BlockSize is the size of bloom filter blocks in bytes.
+	BlockSize = 32
+
+	salt0 = 0x47b6137b
+	salt1 = 0x44974d91
+	salt2 = 0x8824ad5b
+	salt3 = 0xa2b7289d
+	salt4 = 0x705495c7
+	salt5 = 0x2df1424b
+	salt6 = 0x9efc4947
+	salt7 = 0x5c6bfb31
+)
diff --git a/bloom/block_amd64.go b/bloom/block_amd64.go
new file mode 100644
index 0000000..c924a34
--- /dev/null
+++ b/bloom/block_amd64.go
@@ -0,0 +1,36 @@
+//go:build !purego
+
+package bloom
+
+// The functions in this file are SIMD-optimized versions of the functions
+// declared in block_optimized.go for x86 targets.
+//
+// The optimization yields measurable improvements over the pure Go versions:
+//
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/segmentio/parquet-go/bloom
+// cpu: Intel(R) Core(TM) i9-8950HK CPU @ 2.90GHz
+//
+// name         old time/op    new time/op     delta
+// BlockInsert    11.6ns ± 4%      2.0ns ± 3%   -82.37%  (p=0.000 n=8+8)
+// BlockCheck     12.6ns ±28%      2.1ns ± 4%   -83.12%  (p=0.000 n=10+8)
+//
+// name         old speed      new speed       delta
+// BlockInsert  2.73GB/s ±13%  15.70GB/s ± 3%  +475.96%  (p=0.000 n=9+8)
+// BlockCheck   2.59GB/s ±23%  15.06GB/s ± 4%  +482.25%  (p=0.000 n=10+8)
+//
+// Note that the numbers above are a comparison to the routines implemented in
+// block_optimized.go; the delta comparing to functions in block_default.go is
+// significantly larger but not very intersting since those functions have no
+// practical use cases.
+
+//go:noescape
+func blockInsert(b *Block, x uint32)
+
+//go:noescape
+func blockCheck(b *Block, x uint32) bool
+
+func (b *Block) Insert(x uint32) { blockInsert(b, x) }
+
+func (b *Block) Check(x uint32) bool { return blockCheck(b, x) }
diff --git a/bloom/block_amd64.s b/bloom/block_amd64.s
new file mode 100644
index 0000000..b8ade61
--- /dev/null
+++ b/bloom/block_amd64.s
@@ -0,0 +1,70 @@
+//go:build !purego
+
+#include "textflag.h"
+
+DATA ones+0(SB)/4, $1
+DATA ones+4(SB)/4, $1
+DATA ones+8(SB)/4, $1
+DATA ones+12(SB)/4, $1
+DATA ones+16(SB)/4, $1
+DATA ones+20(SB)/4, $1
+DATA ones+24(SB)/4, $1
+DATA ones+28(SB)/4, $1
+GLOBL ones(SB), RODATA|NOPTR, $32
+
+DATA salt+0(SB)/4, $0x47b6137b
+DATA salt+4(SB)/4, $0x44974d91
+DATA salt+8(SB)/4, $0x8824ad5b
+DATA salt+12(SB)/4, $0xa2b7289d
+DATA salt+16(SB)/4, $0x705495c7
+DATA salt+20(SB)/4, $0x2df1424b
+DATA salt+24(SB)/4, $0x9efc4947
+DATA salt+28(SB)/4, $0x5c6bfb31
+GLOBL salt(SB), RODATA|NOPTR, $32
+
+// This initial block is a SIMD implementation of the mask function declared in
+// block_default.go and block_optimized.go. For each of the 8 x 32 bits words of
+// the bloom filter block, the operation performed is:
+//
+//      block[i] = 1 << ((x * salt[i]) >> 27)
+//
+// Arguments
+// ---------
+//
+// * src is a memory location where the value to use when computing the mask is
+//   located. The memory location is not modified.
+//
+// * tmp is a YMM register used as scratch space to hold intermediary results in
+//   the algorithm.
+//
+// * dst is a YMM register where the final mask is written.
+//
+#define generateMask(src, tmp, dst) \
+    VMOVDQA ones(SB), dst \
+    VPBROADCASTD src, tmp \
+    VPMULLD salt(SB), tmp, tmp \
+    VPSRLD $27, tmp, tmp \
+    VPSLLVD tmp, dst, dst
+
+// func blockInsert(b *Block, x uint32)
+TEXT ·blockInsert(SB), NOSPLIT, $0-16
+    MOVQ b+0(FP), AX
+    generateMask(x+8(FP), Y1, Y0)
+    // Set all 1 bits of the mask in the bloom filter block.
+    VPOR (AX), Y0, Y0
+    VMOVDQU Y0, (AX)
+    VZEROUPPER
+    RET
+
+// func blockCheck(b *Block, x uint32) bool
+TEXT ·blockCheck(SB), NOSPLIT, $0-17
+    MOVQ b+0(FP), AX
+    generateMask(x+8(FP), Y1, Y0)
+    // Compare the 1 bits of the mask with the bloom filter block, then compare
+    // the result with the mask, expecting equality if the value `x` was present
+    // in the block.
+    VPAND (AX), Y0, Y1 // Y0 = block & mask
+    VPTEST Y0, Y1      // if (Y0 & ^Y1) != 0 { CF = 1 }
+    SETCS ret+16(FP)   // return CF == 1
+    VZEROUPPER
+    RET
diff --git a/bloom/block_default.go b/bloom/block_default.go
new file mode 100644
index 0000000..016cb73
--- /dev/null
+++ b/bloom/block_default.go
@@ -0,0 +1,65 @@
+//go:build purego && parquet.bloom.no_unroll
+
+package bloom
+
+// This file contains direct translation of the algorithms described in the
+// parquet bloom filter spec:
+// https://github.com/apache/parquet-format/blob/master/BloomFilter.md
+//
+// There are no practical reasons to eable the parquet.bloom.no_unroll build
+// tag, the code is left here as a reference to ensure that the optimized
+// implementations of block operations behave the same as the functions in this
+// file.
+
+var salt = [8]uint32{
+	0: salt0,
+	1: salt1,
+	2: salt2,
+	3: salt3,
+	4: salt4,
+	5: salt5,
+	6: salt6,
+	7: salt7,
+}
+
+func (w *Word) set(i uint) {
+	*w |= Word(1 << i)
+}
+
+func (w Word) has(i uint) bool {
+	return ((w >> Word(i)) & 1) != 0
+}
+
+func mask(x uint32) Block {
+	var b Block
+	for i := uint(0); i < 8; i++ {
+		y := x * salt[i]
+		b[i].set(uint(y) >> 27)
+	}
+	return b
+}
+
+func (b *Block) Insert(x uint32) {
+	masked := mask(x)
+	for i := uint(0); i < 8; i++ {
+		for j := uint(0); j < 32; j++ {
+			if masked[i].has(j) {
+				b[i].set(j)
+			}
+		}
+	}
+}
+
+func (b *Block) Check(x uint32) bool {
+	masked := mask(x)
+	for i := uint(0); i < 8; i++ {
+		for j := uint(0); j < 32; j++ {
+			if masked[i].has(j) {
+				if !b[i].has(j) {
+					return false
+				}
+			}
+		}
+	}
+	return true
+}
diff --git a/bloom/block_optimized.go b/bloom/block_optimized.go
new file mode 100644
index 0000000..e879d15
--- /dev/null
+++ b/bloom/block_optimized.go
@@ -0,0 +1,68 @@
+//go:build (!amd64 || purego) && !parquet.bloom.no_unroll
+
+package bloom
+
+// The functions in this file are optimized versions of the algorithms described
+// in https://github.com/apache/parquet-format/blob/master/BloomFilter.md
+//
+// The functions are manual unrolling of the loops, which yield significant
+// performance improvements:
+//
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/segmentio/parquet-go/bloom
+// cpu: Intel(R) Core(TM) i9-8950HK CPU @ 2.90GHz
+//
+// name         old time/op    new time/op      delta
+// BlockInsert     327ns ± 1%        12ns ± 4%    -96.47%  (p=0.000 n=9+8)
+// BlockCheck      240ns ± 4%        13ns ±28%    -94.75%  (p=0.000 n=8+10)
+//
+// name         old speed      new speed        delta
+// BlockInsert  97.8MB/s ± 1%  2725.0MB/s ±13%  +2686.59%  (p=0.000 n=9+9)
+// BlockCheck    133MB/s ± 4%    2587MB/s ±23%  +1838.46%  (p=0.000 n=8+10)
+//
+// The benchmarks measure throughput based on the byte size of a bloom filter
+// block.
+
+func mask(x uint32) Block {
+	return Block{
+		0: 1 << ((x * salt0) >> 27),
+		1: 1 << ((x * salt1) >> 27),
+		2: 1 << ((x * salt2) >> 27),
+		3: 1 << ((x * salt3) >> 27),
+		4: 1 << ((x * salt4) >> 27),
+		5: 1 << ((x * salt5) >> 27),
+		6: 1 << ((x * salt6) >> 27),
+		7: 1 << ((x * salt7) >> 27),
+	}
+}
+
+func (b *Block) Insert(x uint32) {
+	masked := mask(x)
+	b[0] |= masked[0]
+	b[1] |= masked[1]
+	b[2] |= masked[2]
+	b[3] |= masked[3]
+	b[4] |= masked[4]
+	b[5] |= masked[5]
+	b[6] |= masked[6]
+	b[7] |= masked[7]
+}
+
+func (b *Block) Check(x uint32) bool {
+	masked := mask(x)
+	return ((b[0] & masked[0]) != 0) &&
+		((b[1] & masked[1]) != 0) &&
+		((b[2] & masked[2]) != 0) &&
+		((b[3] & masked[3]) != 0) &&
+		((b[4] & masked[4]) != 0) &&
+		((b[5] & masked[5]) != 0) &&
+		((b[6] & masked[6]) != 0) &&
+		((b[7] & masked[7]) != 0)
+}
+
+func (f SplitBlockFilter) insertBulk(x []uint64) {
+	for i := range x {
+		f.Insert(x[i])
+	}
+}
diff --git a/bloom/bloom.go b/bloom/bloom.go
new file mode 100644
index 0000000..337fe0d
--- /dev/null
+++ b/bloom/bloom.go
@@ -0,0 +1,13 @@
+// Package bloom implements parquet bloom filters.
+package bloom
+
+func fasthash1x64(value uint64, scale int32) uint64 {
+	return ((value >> 32) * uint64(scale)) >> 32
+}
+
+func fasthash4x64(dst, src *[4]uint64, scale int32) {
+	dst[0] = ((src[0] >> 32) * uint64(scale)) >> 32
+	dst[1] = ((src[1] >> 32) * uint64(scale)) >> 32
+	dst[2] = ((src[2] >> 32) * uint64(scale)) >> 32
+	dst[3] = ((src[3] >> 32) * uint64(scale)) >> 32
+}
diff --git a/bloom/filter.go b/bloom/filter.go
new file mode 100644
index 0000000..222462c
--- /dev/null
+++ b/bloom/filter.go
@@ -0,0 +1,105 @@
+package bloom
+
+import (
+	"io"
+	"sync"
+	"unsafe"
+
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+// Filter is an interface representing read-only bloom filters where programs
+// can probe for the possible presence of a hash key.
+type Filter interface {
+	Check(uint64) bool
+}
+
+// MutableFilter is an extension of the Filter interface which supports
+// inserting keys to the filter.
+type MutableFilter interface {
+	Filter
+	Reset()
+	Insert(uint64)
+	InsertBulk([]uint64)
+	Bytes() []byte
+}
+
+// SplitBlockFilter is an in-memory implementation of the parquet bloom filters.
+//
+// This type is useful to construct bloom filters that are later serialized
+// to a storage medium.
+type SplitBlockFilter []Block
+
+// NumSplitBlocksOf returns the number of blocks in a filter intended to hold
+// the given number of values and bits of filter per value.
+//
+// This function is useful to determine the number of blocks when creating bloom
+// filters in memory, for example:
+//
+//	f := make(bloom.SplitBlockFilter, bloom.NumSplitBlocksOf(n, 10))
+//
+func NumSplitBlocksOf(numValues int64, bitsPerValue uint) int {
+	numBytes := bits.ByteCount(uint(numValues) * bitsPerValue)
+	numBlocks := (numBytes + (BlockSize - 1)) / BlockSize
+	return numBlocks
+}
+
+// Reset clears the content of the filter f.
+func (f SplitBlockFilter) Reset() {
+	for i := range f {
+		f[i] = Block{}
+	}
+}
+
+// Block returns a pointer to the block that the given value hashes to in the
+// bloom filter.
+func (f SplitBlockFilter) Block(x uint64) *Block { return &f[fasthash1x64(x, int32(len(f)))] }
+
+// InsertBulk adds all values from x into f.
+func (f SplitBlockFilter) InsertBulk(x []uint64) { filterInsertBulk(f, x) }
+
+// Insert adds x to f.
+func (f SplitBlockFilter) Insert(x uint64) { filterInsert(f, x) }
+
+// Check tests whether x is in f.
+func (f SplitBlockFilter) Check(x uint64) bool { return filterCheck(f, x) }
+
+// Bytes converts f to a byte slice.
+//
+// The returned slice shares the memory of f. The method is intended to be used
+// to serialize the bloom filter to a storage medium.
+func (f SplitBlockFilter) Bytes() []byte {
+	return unsafe.Slice(*(**byte)(unsafe.Pointer(&f)), len(f)*BlockSize)
+}
+
+// CheckSplitBlock is similar to bloom.SplitBlockFilter.Check but reads the
+// bloom filter of n bytes from r.
+//
+// The size n of the bloom filter is assumed to be a multiple of the block size.
+func CheckSplitBlock(r io.ReaderAt, n int64, x uint64) (bool, error) {
+	block := acquireBlock()
+	defer releaseBlock(block)
+	offset := BlockSize * fasthash1x64(x, int32(n/BlockSize))
+	_, err := r.ReadAt(block.Bytes(), int64(offset))
+	return block.Check(uint32(x)), err
+}
+
+var (
+	_ MutableFilter = (SplitBlockFilter)(nil)
+
+	blockPool sync.Pool
+)
+
+func acquireBlock() *Block {
+	b, _ := blockPool.Get().(*Block)
+	if b == nil {
+		b = new(Block)
+	}
+	return b
+}
+
+func releaseBlock(b *Block) {
+	if b != nil {
+		blockPool.Put(b)
+	}
+}
diff --git a/bloom/filter_amd64.go b/bloom/filter_amd64.go
new file mode 100644
index 0000000..3649da5
--- /dev/null
+++ b/bloom/filter_amd64.go
@@ -0,0 +1,33 @@
+//go:build !purego
+
+package bloom
+
+// This file contains the signatures for bloom filter algorithms implemented in
+// filter_amd64.s.
+//
+// The assembly code provides significant speedups on filter inserts and checks,
+// with the greatest gains seen on the bulk insert operation where the use of
+// vectorized code yields great results.
+//
+// The following sections record the kind of performance improvements we were
+// able to measure, comparing with performing the filter block lookups in Go
+// and calling to the block insert and check routines:
+//
+// name              old time/op    new time/op     delta
+// FilterInsertBulk    45.1ns ± 2%    17.8ns ± 3%   -60.41%  (p=0.000 n=10+10)
+// FilterInsert        3.48ns ± 2%     2.55ns ± 1%  -26.86%  (p=0.000 n=10+8)
+// FilterCheck         3.64ns ± 3%     2.66ns ± 2%  -26.82%  (p=0.000 n=10+9)
+//
+// name              old speed      new speed       delta
+// FilterInsertBulk  11.4GB/s ± 2%  28.7GB/s ± 3%  +152.61%  (p=0.000 n=10+10)
+// FilterInsert      9.19GB/s ± 2%  12.56GB/s ± 1%  +36.71%  (p=0.000 n=10+8)
+// FilterCheck       8.80GB/s ± 3%  12.03GB/s ± 2%  +36.61%  (p=0.000 n=10+9)
+
+//go:noescape
+func filterInsertBulk(f []Block, x []uint64)
+
+//go:noescape
+func filterInsert(f []Block, x uint64)
+
+//go:noescape
+func filterCheck(f []Block, x uint64) bool
diff --git a/bloom/filter_amd64.s b/bloom/filter_amd64.s
new file mode 100644
index 0000000..e216210
--- /dev/null
+++ b/bloom/filter_amd64.s
@@ -0,0 +1,129 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// See block_amd64.s for a description of this algorithm.
+#define generateMask(src, dst) \
+    VMOVDQA ones(SB), dst \
+    VPMULLD salt(SB), src, src \
+    VPSRLD $27, src, src \
+    VPSLLVD src, dst, dst
+
+#define applyMask(src, dst) \
+    VPOR dst, src, src \
+    VMOVDQU src, dst
+
+#define fasthash1x64(scale, value) \
+    SHRQ $32, value \
+    IMULQ scale, value \
+    SHRQ $32, value \
+    SHLQ $5, value
+
+#define fasthash4x64(scale, value) \
+    VPSRLQ $32, value, value \
+    VPMULUDQ scale, value, value \
+    VPSRLQ $32, value, value \
+    VPSLLQ $5, value, value
+
+#define extract4x64(srcYMM, srcXMM, tmpXMM, r0, r1, r2, r3) \
+    VEXTRACTI128 $1, srcYMM, tmpXMM \
+    MOVQ srcXMM, r0 \
+    VPEXTRQ $1, srcXMM, r1 \
+    MOVQ tmpXMM, r2 \
+    VPEXTRQ $1, tmpXMM, r3
+
+// func filterInsertBulk(f []Block, x []uint64)
+TEXT ·filterInsertBulk(SB), NOSPLIT, $0-48
+    MOVQ f_base+0(FP), AX
+    MOVQ f_len+8(FP), CX
+    MOVQ x_base+24(FP), BX
+    MOVQ x_len+32(FP), DX
+    VPBROADCASTQ f_base+8(FP), Y0
+
+    // Loop initialization, SI holds the current index in `x`, DI is the number
+    // of elements in `x` rounded down to the nearest multiple of 4.
+    XORQ SI, SI
+    MOVQ DX, DI
+    SHRQ $2, DI
+    SHLQ $2, DI
+loop4x64:
+    CMPQ SI, DI
+    JAE loop
+
+    // The masks and indexes for 4 input hashes are computed in each loop
+    // iteration. The hashes are loaded in Y1 so we can use vector instructions
+    // to compute all 4 indexes in parallel. The lower 32 bits of the hashes are
+    // also broadcasted in 4 YMM registers to compute the 4 masks that will then
+    // be applied to the filter.
+    VMOVDQU (BX)(SI*8), Y1
+    VPBROADCASTD 0(BX)(SI*8), Y2
+    VPBROADCASTD 8(BX)(SI*8), Y3
+    VPBROADCASTD 16(BX)(SI*8), Y4
+    VPBROADCASTD 24(BX)(SI*8), Y5
+
+    fasthash4x64(Y0, Y1)
+    generateMask(Y2, Y6)
+    generateMask(Y3, Y7)
+    generateMask(Y4, Y8)
+    generateMask(Y5, Y9)
+
+    // The next block of instructions move indexes from the vector to general
+    // purpose registers in order to use them as offsets when applying the mask
+    // to the filter.
+    extract4x64(Y1, X1, X10, R8, R9, R10, R11)
+
+    // Apply masks to the filter; this operation is sensitive to aliasing, when
+    // blocks overlap the, CPU has to serialize the reads and writes, which has
+    // a measurable impact on throughput. This would be frequent for small bloom
+    // filters which may have only a few blocks, the probability of seeing
+    // overlapping blocks on large filters should be small enough to make this
+    // a non-issue though.
+    applyMask(Y6, (AX)(R8*1))
+    applyMask(Y7, (AX)(R9*1))
+    applyMask(Y8, (AX)(R10*1))
+    applyMask(Y9, (AX)(R11*1))
+
+    ADDQ $4, SI
+    JMP loop4x64
+loop:
+    // Compute trailing elements in `x` if the length was not a multiple of 4.
+    // This is the same algorithm as the one in the loop4x64 section, working
+    // on a single mask/block pair at a time.
+    CMPQ SI, DX
+    JE done
+    MOVQ (BX)(SI*8), R8
+    VPBROADCASTD (BX)(SI*8), Y0
+    fasthash1x64(CX, R8)
+    generateMask(Y0, Y1)
+    applyMask(Y1, (AX)(R8*1))
+    INCQ SI
+    JMP loop
+done:
+    VZEROUPPER
+    RET
+
+// func filterInsert(f []Block, x uint64)
+TEXT ·filterInsert(SB), NOSPLIT, $0-32
+    MOVQ f_base+0(FP), AX
+    MOVQ f_len+8(FP), BX
+    MOVQ x+24(FP), CX
+    VPBROADCASTD x+24(FP), Y1
+    fasthash1x64(BX, CX)
+    generateMask(Y1, Y0)
+    applyMask(Y0, (AX)(CX*1))
+    VZEROUPPER
+    RET
+
+// func filterCheck(f []Block, x uint64) bool
+TEXT ·filterCheck(SB), NOSPLIT, $0-33
+    MOVQ f_base+0(FP), AX
+    MOVQ f_len+8(FP), BX
+    MOVQ x+24(FP), CX
+    VPBROADCASTD x+24(FP), Y1
+    fasthash1x64(BX, CX)
+    generateMask(Y1, Y0)
+    VPAND (AX)(CX*1), Y0, Y1
+    VPTEST Y0, Y1
+    SETCS ret+32(FP)
+    VZEROUPPER
+    RET
diff --git a/bloom/filter_default.go b/bloom/filter_default.go
new file mode 100644
index 0000000..38fadd5
--- /dev/null
+++ b/bloom/filter_default.go
@@ -0,0 +1,17 @@
+//go:build purego || !amd64
+
+package bloom
+
+func filterInsertBulk(f []Block, x []uint64) {
+	for i := range x {
+		filterInsert(f, x[i])
+	}
+}
+
+func filterInsert(f []Block, x uint64) {
+	f[fasthash1x64(x, int32(len(f)))].Insert(uint32(x))
+}
+
+func filterCheck(f []Block, x uint64) bool {
+	return f[fasthash1x64(x, int32(len(f)))].Check(uint32(x))
+}
diff --git a/bloom/hash.go b/bloom/hash.go
new file mode 100644
index 0000000..c719a85
--- /dev/null
+++ b/bloom/hash.go
@@ -0,0 +1,49 @@
+package bloom
+
+import "github.com/segmentio/parquet-go/bloom/xxhash"
+
+// Hash is an interface abstracting the hashing algorithm used in bloom filters.
+//
+// Hash instances must be safe to use concurrently from multiple goroutines.
+type Hash interface {
+	// Returns the 64 bit hash of the value passed as argument.
+	Sum64(value []byte) uint64
+	// Compute hashes of the array of fixed size values passed as arguments,
+	// returning the number of hashes written to the destination buffer.
+	MultiSum64Uint8(dst []uint64, src []uint8) int
+	MultiSum64Uint16(dst []uint64, src []uint16) int
+	MultiSum64Uint32(dst []uint64, src []uint32) int
+	MultiSum64Uint64(dst []uint64, src []uint64) int
+	MultiSum64Uint128(dst []uint64, src [][16]byte) int
+}
+
+// XXH64 is an implementation fo the Hash interface using the XXH64 algorithm.
+type XXH64 struct{}
+
+func (XXH64) Sum64(b []byte) uint64 {
+	return xxhash.Sum64(b)
+}
+
+func (XXH64) MultiSum64Uint8(h []uint64, v []uint8) int {
+	return xxhash.MultiSum64Uint8(h, v)
+}
+
+func (XXH64) MultiSum64Uint16(h []uint64, v []uint16) int {
+	return xxhash.MultiSum64Uint16(h, v)
+}
+
+func (XXH64) MultiSum64Uint32(h []uint64, v []uint32) int {
+	return xxhash.MultiSum64Uint32(h, v)
+}
+
+func (XXH64) MultiSum64Uint64(h []uint64, v []uint64) int {
+	return xxhash.MultiSum64Uint64(h, v)
+}
+
+func (XXH64) MultiSum64Uint128(h []uint64, v [][16]byte) int {
+	return xxhash.MultiSum64Uint128(h, v)
+}
+
+var (
+	_ Hash = XXH64{}
+)
diff --git a/bloom/xxhash/LICENSE b/bloom/xxhash/LICENSE
new file mode 100644
index 0000000..80bef2e
--- /dev/null
+++ b/bloom/xxhash/LICENSE
@@ -0,0 +1,27 @@
+The following files in this directory were derived from the open-source
+project at https://github.com/cespare/xxhash. A copy of the original
+license is provided below.
+------------------------------------------------------------------------
+
+Copyright (c) 2016 Caleb Spare
+
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/bloom/xxhash/sum64uint.go b/bloom/xxhash/sum64uint.go
new file mode 100644
index 0000000..84423d0
--- /dev/null
+++ b/bloom/xxhash/sum64uint.go
@@ -0,0 +1,37 @@
+package xxhash
+
+func Sum64Uint8(v uint8) uint64 {
+	h := prime5 + 1
+	h ^= uint64(v) * prime5
+	return avalanche(rol11(h) * prime1)
+}
+
+func Sum64Uint16(v uint16) uint64 {
+	h := prime5 + 2
+	h ^= uint64(v&0xFF) * prime5
+	h = rol11(h) * prime1
+	h ^= uint64(v>>8) * prime5
+	h = rol11(h) * prime1
+	return avalanche(h)
+}
+
+func Sum64Uint32(v uint32) uint64 {
+	h := prime5 + 4
+	h ^= uint64(v) * prime1
+	return avalanche(rol23(h)*prime2 + prime3)
+}
+
+func Sum64Uint64(v uint64) uint64 {
+	h := prime5 + 8
+	h ^= round(0, v)
+	return avalanche(rol27(h)*prime1 + prime4)
+}
+
+func Sum64Uint128(v [16]byte) uint64 {
+	h := prime5 + 16
+	h ^= round(0, u64(v[:8]))
+	h = rol27(h)*prime1 + prime4
+	h ^= round(0, u64(v[8:]))
+	h = rol27(h)*prime1 + prime4
+	return avalanche(h)
+}
diff --git a/bloom/xxhash/sum64uint_amd64.go b/bloom/xxhash/sum64uint_amd64.go
new file mode 100644
index 0000000..b70ae8c
--- /dev/null
+++ b/bloom/xxhash/sum64uint_amd64.go
@@ -0,0 +1,43 @@
+//go:build !purego
+
+package xxhash
+
+// This file contains the declaration of signatures for the multi hashing
+// functions implemented in sum64uint_amd64.s, which provides vectorized
+// versions of the algorithms written in sum64uint_purego.go.
+//
+// The use of SIMD optimization yields measurable throughput increases when
+// computing multiple hash values in parallel compared to hashing values
+// individually in loops:
+//
+// name                   old speed      new speed      delta
+// MultiSum64Uint8/4KB    4.94GB/s ± 2%  6.82GB/s ± 5%  +38.00%  (p=0.000 n=10+10)
+// MultiSum64Uint16/4KB   3.44GB/s ± 2%  4.63GB/s ± 4%  +34.56%  (p=0.000 n=10+10)
+// MultiSum64Uint32/4KB   4.84GB/s ± 2%  6.39GB/s ± 4%  +31.94%  (p=0.000 n=10+10)
+// MultiSum64Uint64/4KB   3.77GB/s ± 2%  4.95GB/s ± 2%  +31.14%  (p=0.000 n=9+10)
+// MultiSum64Uint128/4KB  1.84GB/s ± 2%  3.11GB/s ± 4%  +68.70%  (p=0.000 n=9+10)
+//
+// name                   old hash/s     new hash/s     delta
+// MultiSum64Uint8/4KB        617M ± 2%      852M ± 5%  +38.00%  (p=0.000 n=10+10)
+// MultiSum64Uint16/4KB       431M ± 2%      579M ± 4%  +34.56%  (p=0.000 n=10+10)
+// MultiSum64Uint32/4KB       605M ± 2%      799M ± 4%  +31.94%  (p=0.000 n=10+10)
+// MultiSum64Uint64/4KB       471M ± 2%      618M ± 2%  +31.14%  (p=0.000 n=9+10)
+// MultiSum64Uint128/4KB      231M ± 2%      389M ± 4%  +68.70%  (p=0.000 n=9+10)
+//
+// The benchmarks measure the throughput of hashes produced, as a rate of values
+// and bytes.
+
+//go:noescape
+func MultiSum64Uint8(h []uint64, v []uint8) int
+
+//go:noescape
+func MultiSum64Uint16(h []uint64, v []uint16) int
+
+//go:noescape
+func MultiSum64Uint32(h []uint64, v []uint32) int
+
+//go:noescape
+func MultiSum64Uint64(h []uint64, v []uint64) int
+
+//go:noescape
+func MultiSum64Uint128(h []uint64, v [][16]byte) int
diff --git a/bloom/xxhash/sum64uint_amd64.s b/bloom/xxhash/sum64uint_amd64.s
new file mode 100644
index 0000000..2a2e140
--- /dev/null
+++ b/bloom/xxhash/sum64uint_amd64.s
@@ -0,0 +1,233 @@
+//go:build !purego
+
+#include "textflag.h"
+
+#define PRIME1 0x9E3779B185EBCA87
+#define PRIME2 0xC2B2AE3D27D4EB4F
+#define PRIME3 0x165667B19E3779F9
+#define PRIME4 0x85EBCA77C2B2AE63
+#define PRIME5 0x27D4EB2F165667C5
+
+#define prime1 R12
+#define prime2 R13
+#define prime3 R14
+#define prime4 R15
+#define prime5 R15 // same as prime4 because they are not used together
+
+#define round(input, acc) \
+	IMULQ prime2, input \
+	ADDQ  input, acc \
+	ROLQ  $31, acc \
+	IMULQ prime1, acc
+
+#define avalanche(tmp, acc) \
+    MOVQ acc, tmp \
+    SHRQ $33, tmp \
+    XORQ tmp, acc \
+    IMULQ prime2, acc \
+    MOVQ acc, tmp \
+    SHRQ $29, tmp \
+    XORQ tmp, acc \
+    IMULQ prime3, acc \
+    MOVQ acc, tmp \
+    SHRQ $32, tmp \
+    XORQ tmp, acc
+
+// func MultiSum64Uint8(h []uint64, v []uint8) int
+TEXT ·MultiSum64Uint8(SB), NOSPLIT, $0-54
+    MOVQ $PRIME1, prime1
+    MOVQ $PRIME2, prime2
+    MOVQ $PRIME3, prime3
+    MOVQ $PRIME5, prime5
+
+    MOVQ h_base+0(FP), AX
+    MOVQ h_len+8(FP), CX
+    MOVQ v_base+24(FP), BX
+    MOVQ v_len+32(FP), DX
+
+    CMPQ CX, DX
+    CMOVQGT DX, CX
+    MOVQ CX, ret+48(FP)
+
+    XORQ SI, SI
+loop:
+    CMPQ SI, CX
+    JE done
+    MOVQ $PRIME5+1, R8
+    MOVBQZX (BX)(SI*1), R9
+
+    IMULQ prime5, R9
+    XORQ R9, R8
+    ROLQ $11, R8
+    IMULQ prime1, R8
+    avalanche(R9, R8)
+
+    MOVQ R8, (AX)(SI*8)
+    INCQ SI
+    JMP loop
+done:
+    RET
+
+// func MultiSum64Uint16(h []uint64, v []uint16) int
+TEXT ·MultiSum64Uint16(SB), NOSPLIT, $0-54
+    MOVQ $PRIME1, prime1
+    MOVQ $PRIME2, prime2
+    MOVQ $PRIME3, prime3
+    MOVQ $PRIME5, prime5
+
+    MOVQ h_base+0(FP), AX
+    MOVQ h_len+8(FP), CX
+    MOVQ v_base+24(FP), BX
+    MOVQ v_len+32(FP), DX
+
+    CMPQ CX, DX
+    CMOVQGT DX, CX
+    MOVQ CX, ret+48(FP)
+
+    XORQ SI, SI
+loop:
+    CMPQ SI, CX
+    JE done
+    MOVQ $PRIME5+2, R8
+    MOVWQZX (BX)(SI*2), R9
+
+    MOVQ R9, R10
+    SHRQ $8, R10
+    ANDQ $0xFF, R9
+
+    IMULQ prime5, R9
+    XORQ R9, R8
+    ROLQ $11, R8
+    IMULQ prime1, R8
+
+    IMULQ prime5, R10
+    XORQ R10, R8
+    ROLQ $11, R8
+    IMULQ prime1, R8
+
+    avalanche(R9, R8)
+
+    MOVQ R8, (AX)(SI*8)
+    INCQ SI
+    JMP loop
+done:
+    RET
+
+// func MultiSum64Uint32(h []uint64, v []uint32) int
+TEXT ·MultiSum64Uint32(SB), NOSPLIT, $0-54
+    MOVQ $PRIME1, prime1
+    MOVQ $PRIME2, prime2
+    MOVQ $PRIME3, prime3
+
+    MOVQ h_base+0(FP), AX
+    MOVQ h_len+8(FP), CX
+    MOVQ v_base+24(FP), BX
+    MOVQ v_len+32(FP), DX
+
+    CMPQ CX, DX
+    CMOVQGT DX, CX
+    MOVQ CX, ret+48(FP)
+
+    XORQ SI, SI
+loop:
+    CMPQ SI, CX
+    JE done
+    MOVQ $PRIME5+4, R8
+    MOVLQZX (BX)(SI*4), R9
+
+    IMULQ prime1, R9
+    XORQ R9, R8
+    ROLQ $23, R8
+    IMULQ prime2, R8
+    ADDQ prime3, R8
+    avalanche(R9, R8)
+
+    MOVQ R8, (AX)(SI*8)
+    INCQ SI
+    JMP loop
+done:
+    RET
+
+// func MultiSum64Uint64(h []uint64, v []uint64) int
+TEXT ·MultiSum64Uint64(SB), NOSPLIT, $0-54
+    MOVQ $PRIME1, prime1
+    MOVQ $PRIME2, prime2
+    MOVQ $PRIME3, prime3
+    MOVQ $PRIME4, prime4
+
+    MOVQ h_base+0(FP), AX
+    MOVQ h_len+8(FP), CX
+    MOVQ v_base+24(FP), BX
+    MOVQ v_len+32(FP), DX
+
+    CMPQ CX, DX
+    CMOVQGT DX, CX
+    MOVQ CX, ret+48(FP)
+
+    XORQ SI, SI
+loop:
+    CMPQ SI, CX
+    JE done
+    MOVQ $PRIME5+8, R8
+    MOVQ (BX)(SI*8), R9
+
+    XORQ R10, R10
+    round(R9, R10)
+    XORQ R10, R8
+    ROLQ $27, R8
+    IMULQ prime1, R8
+    ADDQ prime4, R8
+    avalanche(R9, R8)
+
+    MOVQ R8, (AX)(SI*8)
+    INCQ SI
+    JMP loop
+done:
+    RET
+
+// func MultiSum64Uint128(h []uint64, v [][16]byte) int
+TEXT ·MultiSum64Uint128(SB), NOSPLIT, $0-54
+    MOVQ $PRIME1, prime1
+    MOVQ $PRIME2, prime2
+    MOVQ $PRIME3, prime3
+    MOVQ $PRIME4, prime4
+
+    MOVQ h_base+0(FP), AX
+    MOVQ h_len+8(FP), CX
+    MOVQ v_base+24(FP), BX
+    MOVQ v_len+32(FP), DX
+
+    CMPQ CX, DX
+    CMOVQGT DX, CX
+    MOVQ CX, ret+48(FP)
+
+    XORQ SI, SI
+loop:
+    CMPQ SI, CX
+    JE done
+    MOVQ $PRIME5+16, R8
+    MOVQ (BX), DX
+    MOVQ 8(BX), DI
+
+    XORQ R9, R9
+    XORQ R10, R10
+    round(DX, R9)
+    round(DI, R10)
+
+    XORQ R9, R8
+    ROLQ $27, R8
+    IMULQ prime1, R8
+    ADDQ prime4, R8
+
+    XORQ R10, R8
+    ROLQ $27, R8
+    IMULQ prime1, R8
+    ADDQ prime4, R8
+
+    avalanche(R9, R8)
+    MOVQ R8, (AX)(SI*8)
+    ADDQ $16, BX
+    INCQ SI
+    JMP loop
+done:
+    RET
diff --git a/bloom/xxhash/sum64uint_purego.go b/bloom/xxhash/sum64uint_purego.go
new file mode 100644
index 0000000..d4606de
--- /dev/null
+++ b/bloom/xxhash/sum64uint_purego.go
@@ -0,0 +1,60 @@
+//go:build purego || !amd64
+
+package xxhash
+
+func MultiSum64Uint8(h []uint64, v []uint8) int {
+	n := min(len(h), len(v))
+	h = h[:n]
+	v = v[:n]
+	for i := range v {
+		h[i] = Sum64Uint8(v[i])
+	}
+	return n
+}
+
+func MultiSum64Uint16(h []uint64, v []uint16) int {
+	n := min(len(h), len(v))
+	h = h[:n]
+	v = v[:n]
+	for i := range v {
+		h[i] = Sum64Uint16(v[i])
+	}
+	return n
+}
+
+func MultiSum64Uint32(h []uint64, v []uint32) int {
+	n := min(len(h), len(v))
+	h = h[:n]
+	v = v[:n]
+	for i := range v {
+		h[i] = Sum64Uint32(v[i])
+	}
+	return n
+}
+
+func MultiSum64Uint64(h []uint64, v []uint64) int {
+	n := min(len(h), len(v))
+	h = h[:n]
+	v = v[:n]
+	for i := range v {
+		h[i] = Sum64Uint64(v[i])
+	}
+	return n
+}
+
+func MultiSum64Uint128(h []uint64, v [][16]byte) int {
+	n := min(len(h), len(v))
+	h = h[:n]
+	v = v[:n]
+	for i := range v {
+		h[i] = Sum64Uint128(v[i])
+	}
+	return n
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
diff --git a/bloom/xxhash/xxhash.go b/bloom/xxhash/xxhash.go
new file mode 100644
index 0000000..348ce4d
--- /dev/null
+++ b/bloom/xxhash/xxhash.go
@@ -0,0 +1,55 @@
+// Package xxhash is an extension of github.com/cespare/xxhash which adds
+// routines optimized to hash arrays of fixed size elements.
+package xxhash
+
+import (
+	"encoding/binary"
+	"math/bits"
+)
+
+const (
+	prime1 uint64 = 0x9E3779B185EBCA87
+	prime2 uint64 = 0xC2B2AE3D27D4EB4F
+	prime3 uint64 = 0x165667B19E3779F9
+	prime4 uint64 = 0x85EBCA77C2B2AE63
+	prime5 uint64 = 0x27D4EB2F165667C5
+	// Precomputed operations because the compiler otherwise complains that the
+	// results overflow 64 bit integers.
+	prime1plus2 uint64 = 0x60EA27EEADC0B5D6 // prime1 + prime2
+	negprime1   uint64 = 0x61C8864E7A143579 // -prime1
+)
+
+func avalanche(h uint64) uint64 {
+	h ^= h >> 33
+	h *= prime2
+	h ^= h >> 29
+	h *= prime3
+	h ^= h >> 32
+	return h
+}
+
+func round(acc, input uint64) uint64 {
+	acc += input * prime2
+	acc = rol31(acc)
+	acc *= prime1
+	return acc
+}
+
+func mergeRound(acc, val uint64) uint64 {
+	val = round(0, val)
+	acc ^= val
+	acc = acc*prime1 + prime4
+	return acc
+}
+
+func u64(b []byte) uint64 { return binary.LittleEndian.Uint64(b) }
+func u32(b []byte) uint32 { return binary.LittleEndian.Uint32(b) }
+
+func rol1(x uint64) uint64  { return bits.RotateLeft64(x, 1) }
+func rol7(x uint64) uint64  { return bits.RotateLeft64(x, 7) }
+func rol11(x uint64) uint64 { return bits.RotateLeft64(x, 11) }
+func rol12(x uint64) uint64 { return bits.RotateLeft64(x, 12) }
+func rol18(x uint64) uint64 { return bits.RotateLeft64(x, 18) }
+func rol23(x uint64) uint64 { return bits.RotateLeft64(x, 23) }
+func rol27(x uint64) uint64 { return bits.RotateLeft64(x, 27) }
+func rol31(x uint64) uint64 { return bits.RotateLeft64(x, 31) }
diff --git a/bloom/xxhash/xxhash_amd64.go b/bloom/xxhash/xxhash_amd64.go
new file mode 100644
index 0000000..4b32923
--- /dev/null
+++ b/bloom/xxhash/xxhash_amd64.go
@@ -0,0 +1,6 @@
+//go:build !purego
+
+package xxhash
+
+// Sum64 computes the 64-bit xxHash digest of b.
+func Sum64(b []byte) uint64
diff --git a/bloom/xxhash/xxhash_amd64.s b/bloom/xxhash/xxhash_amd64.s
new file mode 100644
index 0000000..8f88dc1
--- /dev/null
+++ b/bloom/xxhash/xxhash_amd64.s
@@ -0,0 +1,180 @@
+//go:build !purego
+
+#include "textflag.h"
+
+#define PRIME1 0x9E3779B185EBCA87
+#define PRIME2 0xC2B2AE3D27D4EB4F
+#define PRIME3 0x165667B19E3779F9
+#define PRIME4 0x85EBCA77C2B2AE63
+#define PRIME5 0x27D4EB2F165667C5
+
+DATA prime3<>+0(SB)/8, $PRIME3
+GLOBL prime3<>(SB), RODATA|NOPTR, $8
+
+DATA prime5<>+0(SB)/8, $PRIME5
+GLOBL prime5<>(SB), RODATA|NOPTR, $8
+
+// Register allocation:
+// AX	h
+// SI	pointer to advance through b
+// DX	n
+// BX	loop end
+// R8	v1, k1
+// R9	v2
+// R10	v3
+// R11	v4
+// R12	tmp
+// R13	PRIME1
+// R14	PRIME2
+// DI	PRIME4
+
+// round reads from and advances the buffer pointer in SI.
+// It assumes that R13 has PRIME1 and R14 has PRIME2.
+#define round(r) \
+	MOVQ  (SI), R12 \
+	ADDQ  $8, SI    \
+	IMULQ R14, R12  \
+	ADDQ  R12, r    \
+	ROLQ  $31, r    \
+	IMULQ R13, r
+
+// mergeRound applies a merge round on the two registers acc and val.
+// It assumes that R13 has PRIME1, R14 has PRIME2, and DI has PRIME4.
+#define mergeRound(acc, val) \
+	IMULQ R14, val \
+	ROLQ  $31, val \
+	IMULQ R13, val \
+	XORQ  val, acc \
+	IMULQ R13, acc \
+	ADDQ  DI, acc
+
+// func Sum64(b []byte) uint64
+TEXT ·Sum64(SB), NOSPLIT, $0-32
+	// Load fixed primes.
+	MOVQ $PRIME1, R13
+	MOVQ $PRIME2, R14
+	MOVQ $PRIME4, DI
+
+	// Load slice.
+	MOVQ b_base+0(FP), SI
+	MOVQ b_len+8(FP), DX
+	LEAQ (SI)(DX*1), BX
+
+	// The first loop limit will be len(b)-32.
+	SUBQ $32, BX
+
+	// Check whether we have at least one block.
+	CMPQ DX, $32
+	JLT  noBlocks
+
+	// Set up initial state (v1, v2, v3, v4).
+	MOVQ R13, R8
+	ADDQ R14, R8
+	MOVQ R14, R9
+	XORQ R10, R10
+	XORQ R11, R11
+	SUBQ R13, R11
+
+	// Loop until SI > BX.
+blockLoop:
+	round(R8)
+	round(R9)
+	round(R10)
+	round(R11)
+
+	CMPQ SI, BX
+	JLE  blockLoop
+
+	MOVQ R8, AX
+	ROLQ $1, AX
+	MOVQ R9, R12
+	ROLQ $7, R12
+	ADDQ R12, AX
+	MOVQ R10, R12
+	ROLQ $12, R12
+	ADDQ R12, AX
+	MOVQ R11, R12
+	ROLQ $18, R12
+	ADDQ R12, AX
+
+	mergeRound(AX, R8)
+	mergeRound(AX, R9)
+	mergeRound(AX, R10)
+	mergeRound(AX, R11)
+
+	JMP afterBlocks
+
+noBlocks:
+	MOVQ $PRIME5, AX
+
+afterBlocks:
+	ADDQ DX, AX
+
+	// Right now BX has len(b)-32, and we want to loop until SI > len(b)-8.
+	ADDQ $24, BX
+
+	CMPQ SI, BX
+	JG   fourByte
+
+wordLoop:
+	// Calculate k1.
+	MOVQ  (SI), R8
+	ADDQ  $8, SI
+	IMULQ R14, R8
+	ROLQ  $31, R8
+	IMULQ R13, R8
+
+	XORQ  R8, AX
+	ROLQ  $27, AX
+	IMULQ R13, AX
+	ADDQ  DI, AX
+
+	CMPQ SI, BX
+	JLE  wordLoop
+
+fourByte:
+	ADDQ $4, BX
+	CMPQ SI, BX
+	JG   singles
+
+	MOVL  (SI), R8
+	ADDQ  $4, SI
+	IMULQ R13, R8
+	XORQ  R8, AX
+
+	ROLQ  $23, AX
+	IMULQ R14, AX
+	ADDQ  prime3<>(SB), AX
+
+singles:
+	ADDQ $4, BX
+	CMPQ SI, BX
+	JGE  finalize
+
+singlesLoop:
+	MOVBQZX (SI), R12
+	ADDQ    $1, SI
+	IMULQ   prime5<>(SB), R12
+	XORQ    R12, AX
+
+	ROLQ  $11, AX
+	IMULQ R13, AX
+
+	CMPQ SI, BX
+	JL   singlesLoop
+
+finalize:
+	MOVQ  AX, R12
+	SHRQ  $33, R12
+	XORQ  R12, AX
+	IMULQ R14, AX
+	MOVQ  AX, R12
+	SHRQ  $29, R12
+	XORQ  R12, AX
+	IMULQ prime3<>(SB), AX
+	MOVQ  AX, R12
+	SHRQ  $32, R12
+	XORQ  R12, AX
+
+	MOVQ AX, ret+24(FP)
+	RET
diff --git a/bloom/xxhash/xxhash_purego.go b/bloom/xxhash/xxhash_purego.go
new file mode 100644
index 0000000..7d44b8c
--- /dev/null
+++ b/bloom/xxhash/xxhash_purego.go
@@ -0,0 +1,50 @@
+//go:build purego
+
+package xxhash
+
+// Sum64 computes the 64-bit xxHash digest of b.
+func Sum64(b []byte) uint64 {
+	var n = len(b)
+	var h uint64
+
+	if n >= 32 {
+		v1 := prime1plus2
+		v2 := prime2
+		v3 := uint64(0)
+		v4 := negprime1
+		for len(b) >= 32 {
+			v1 = round(v1, u64(b[0:8:len(b)]))
+			v2 = round(v2, u64(b[8:16:len(b)]))
+			v3 = round(v3, u64(b[16:24:len(b)]))
+			v4 = round(v4, u64(b[24:32:len(b)]))
+			b = b[32:len(b):len(b)]
+		}
+		h = rol1(v1) + rol7(v2) + rol12(v3) + rol18(v4)
+		h = mergeRound(h, v1)
+		h = mergeRound(h, v2)
+		h = mergeRound(h, v3)
+		h = mergeRound(h, v4)
+	} else {
+		h = prime5
+	}
+
+	h += uint64(n)
+
+	i, end := 0, len(b)
+	for ; i+8 <= end; i += 8 {
+		k1 := round(0, u64(b[i:i+8:len(b)]))
+		h ^= k1
+		h = rol27(h)*prime1 + prime4
+	}
+	if i+4 <= end {
+		h ^= uint64(u32(b[i:i+4:len(b)])) * prime1
+		h = rol23(h)*prime2 + prime3
+		i += 4
+	}
+	for ; i < end; i++ {
+		h ^= uint64(b[i]) * prime5
+		h = rol11(h) * prime1
+	}
+
+	return avalanche(h)
+}
diff --git a/column_buffer.go b/column_buffer.go
index 7ccfe93..f2049c2 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -152,6 +152,10 @@ func (col *optionalColumnBuffer) Type() Type {
 	return col.base.Type()
 }
 
+func (col *optionalColumnBuffer) NumValues() int64 {
+	return int64(len(col.definitionLevels))
+}
+
 func (col *optionalColumnBuffer) ColumnIndex() ColumnIndex {
 	return columnIndexOfNullable(col.base, col.maxDefinitionLevel, col.definitionLevels)
 }
@@ -160,6 +164,10 @@ func (col *optionalColumnBuffer) OffsetIndex() OffsetIndex {
 	return col.base.OffsetIndex()
 }
 
+func (col *optionalColumnBuffer) BloomFilter() BloomFilter {
+	return col.base.BloomFilter()
+}
+
 func (col *optionalColumnBuffer) Dictionary() Dictionary {
 	return col.base.Dictionary()
 }
@@ -394,6 +402,10 @@ func (col *repeatedColumnBuffer) Type() Type {
 	return col.base.Type()
 }
 
+func (col *repeatedColumnBuffer) NumValues() int64 {
+	return int64(len(col.definitionLevels))
+}
+
 func (col *repeatedColumnBuffer) ColumnIndex() ColumnIndex {
 	return columnIndexOfNullable(col.base, col.maxDefinitionLevel, col.definitionLevels)
 }
@@ -402,6 +414,10 @@ func (col *repeatedColumnBuffer) OffsetIndex() OffsetIndex {
 	return col.base.OffsetIndex()
 }
 
+func (col *repeatedColumnBuffer) BloomFilter() BloomFilter {
+	return col.base.BloomFilter()
+}
+
 func (col *repeatedColumnBuffer) Dictionary() Dictionary {
 	return col.base.Dictionary()
 }
@@ -688,6 +704,8 @@ func (col *booleanColumnBuffer) ColumnIndex() ColumnIndex { return booleanPageIn
 
 func (col *booleanColumnBuffer) OffsetIndex() OffsetIndex { return booleanPageIndex{&col.booleanPage} }
 
+func (col *booleanColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *booleanColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *booleanColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -770,6 +788,8 @@ func (col *int32ColumnBuffer) ColumnIndex() ColumnIndex { return int32PageIndex{
 
 func (col *int32ColumnBuffer) OffsetIndex() OffsetIndex { return int32PageIndex{&col.int32Page} }
 
+func (col *int32ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *int32ColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *int32ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -850,6 +870,8 @@ func (col *int64ColumnBuffer) ColumnIndex() ColumnIndex { return int64PageIndex{
 
 func (col *int64ColumnBuffer) OffsetIndex() OffsetIndex { return int64PageIndex{&col.int64Page} }
 
+func (col *int64ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *int64ColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *int64ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -930,6 +952,8 @@ func (col *int96ColumnBuffer) ColumnIndex() ColumnIndex { return int96PageIndex{
 
 func (col *int96ColumnBuffer) OffsetIndex() OffsetIndex { return int96PageIndex{&col.int96Page} }
 
+func (col *int96ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *int96ColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *int96ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -1010,6 +1034,8 @@ func (col *floatColumnBuffer) ColumnIndex() ColumnIndex { return floatPageIndex{
 
 func (col *floatColumnBuffer) OffsetIndex() OffsetIndex { return floatPageIndex{&col.floatPage} }
 
+func (col *floatColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *floatColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *floatColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -1090,6 +1116,8 @@ func (col *doubleColumnBuffer) ColumnIndex() ColumnIndex { return doublePageInde
 
 func (col *doubleColumnBuffer) OffsetIndex() OffsetIndex { return doublePageIndex{&col.doublePage} }
 
+func (col *doubleColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *doubleColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *doubleColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -1174,6 +1202,8 @@ func (col *byteArrayColumnBuffer) OffsetIndex() OffsetIndex {
 	return byteArrayPageIndex{&col.byteArrayPage}
 }
 
+func (col *byteArrayColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *byteArrayColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *byteArrayColumnBuffer) Pages() Pages { return onePage(col.Page()) }
@@ -1262,6 +1292,8 @@ func (col *fixedLenByteArrayColumnBuffer) OffsetIndex() OffsetIndex {
 	return fixedLenByteArrayPageIndex{&col.fixedLenByteArrayPage}
 }
 
+func (col *fixedLenByteArrayColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *fixedLenByteArrayColumnBuffer) Dictionary() Dictionary { return nil }
 
 func (col *fixedLenByteArrayColumnBuffer) Pages() Pages { return onePage(col.Page()) }
diff --git a/column_chunk.go b/column_chunk.go
index 9ca0476..98324f3 100644
--- a/column_chunk.go
+++ b/column_chunk.go
@@ -25,6 +25,13 @@ type ColumnChunk interface {
 	// If the column chunk does not have a page index, the methods return nil.
 	ColumnIndex() ColumnIndex
 	OffsetIndex() OffsetIndex
+	BloomFilter() BloomFilter
+
+	// Returns the number of values in the column chunk.
+	//
+	// This quantity may differ from the number of rows in the parent row group
+	// because repeated columns may hold zero or more values per row.
+	NumValues() int64
 }
 
 // Pages is an interface implemented by page readers returned by calling the
diff --git a/concat.go b/concat.go
index 2545dbe..e5d3393 100644
--- a/concat.go
+++ b/concat.go
@@ -62,6 +62,14 @@ func (c *concatenatedColumnChunk) Type() Type {
 	return nil
 }
 
+func (c *concatenatedColumnChunk) NumValues() int64 {
+	n := int64(0)
+	for i := range c.chunks {
+		n += c.chunks[i].NumValues()
+	}
+	return n
+}
+
 func (c *concatenatedColumnChunk) Column() int {
 	return c.column
 }
@@ -71,23 +79,83 @@ func (c *concatenatedColumnChunk) Pages() Pages {
 }
 
 func (c *concatenatedColumnChunk) ColumnIndex() ColumnIndex {
-	// TODO: changin the ColumnIndex type from a concrete type to an interface
-	// means that we could create a concatenated view of the indexes instead of
-	// having to reallocate memory buffers.
+	// TODO: implement
 	return nil
 }
 
 func (c *concatenatedColumnChunk) OffsetIndex() OffsetIndex {
-	// TODO: we cannot really reconstruct the offsets here because we do not
-	// know whether the parent row groups belong to the same file.
-	//
-	// Here as well, changing the OffsetIndex type to an interface could let us
-	// embed useful information to map the index back to the original chunk and
-	// allow leveraging it to lookup pages, even if there are no absolute file
-	// offset.
+	// TODO: implement
 	return nil
 }
 
+func (c *concatenatedColumnChunk) BloomFilter() BloomFilter {
+	return concatenatedBloomFilter{c}
+}
+
+type concatenatedBloomFilter struct{ *concatenatedColumnChunk }
+
+func (f concatenatedBloomFilter) ReadAt(b []byte, off int64) (int, error) {
+	// TODO: add a test for this function
+	i := 0
+
+	for i < len(f.chunks) {
+		if r := f.chunks[i].BloomFilter(); r != nil {
+			size := r.Size()
+			if off < size {
+				break
+			}
+			off -= size
+		}
+		i++
+	}
+
+	if i == len(f.chunks) {
+		return 0, io.EOF
+	}
+
+	rn := int(0)
+	for len(b) > 0 {
+		if r := f.chunks[i].BloomFilter(); r != nil {
+			n, err := r.ReadAt(b, off)
+			rn += n
+			if err != nil {
+				return rn, err
+			}
+			if b = b[n:]; len(b) == 0 {
+				return rn, nil
+			}
+			off += int64(n)
+		}
+		i++
+	}
+
+	if i == len(f.chunks) {
+		return rn, io.EOF
+	}
+	return rn, nil
+}
+
+func (f concatenatedBloomFilter) Size() int64 {
+	size := int64(0)
+	for _, c := range f.chunks {
+		if b := c.BloomFilter(); b != nil {
+			size += b.Size()
+		}
+	}
+	return size
+}
+
+func (f concatenatedBloomFilter) Check(key []byte) (bool, error) {
+	for _, c := range f.chunks {
+		if b := c.BloomFilter(); b != nil {
+			if ok, err := b.Check(key); ok || err != nil {
+				return ok, err
+			}
+		}
+	}
+	return false, nil
+}
+
 type concatenatedPages struct {
 	pages  Pages
 	index  int
diff --git a/config.go b/config.go
index 8776369..cec261e 100644
--- a/config.go
+++ b/config.go
@@ -13,6 +13,7 @@ const (
 	DefaultDataPageVersion      = 2
 	DefaultDataPageStatistics   = false
 	DefaultSkipPageIndex        = false
+	DefaultSkipBloomFilters     = false
 )
 
 // The FileConfig type carries configuration options for parquet files.
@@ -21,18 +22,21 @@ const (
 // as argument to the OpenFile function when needed, for example:
 //
 //	f, err := parquet.OpenFile(reader, size, &parquet.FileConfig{
-//		SkipPageIndex: true,
+//		SkipPageIndex:    true,
+//		SkipBloomFilters: true,
 //	})
 //
 type FileConfig struct {
-	SkipPageIndex bool
+	SkipPageIndex    bool
+	SkipBloomFilters bool
 }
 
 // DefaultFileConfig returns a new FileConfig value initialized with the
 // default file configuration.
 func DefaultFileConfig() *FileConfig {
 	return &FileConfig{
-		SkipPageIndex: DefaultSkipPageIndex,
+		SkipPageIndex:    DefaultSkipPageIndex,
+		SkipBloomFilters: DefaultSkipBloomFilters,
 	}
 }
 
@@ -57,7 +61,8 @@ func (c *FileConfig) Apply(options ...FileOption) {
 // ConfigureFile applies configuration options from c to config.
 func (c *FileConfig) ConfigureFile(config *FileConfig) {
 	*config = FileConfig{
-		SkipPageIndex: config.SkipPageIndex,
+		SkipPageIndex:    config.SkipPageIndex,
+		SkipBloomFilters: config.SkipBloomFilters,
 	}
 }
 
@@ -130,6 +135,7 @@ type WriterConfig struct {
 	DataPageStatistics   bool
 	KeyValueMetadata     map[string]string
 	Schema               *Schema
+	BloomFilters         []BloomFilterColumn
 }
 
 // DefaultWriterConfig returns a new WriterConfig value initialized with the
@@ -183,6 +189,7 @@ func (c *WriterConfig) ConfigureWriter(config *WriterConfig) {
 		DataPageStatistics:   config.DataPageStatistics,
 		KeyValueMetadata:     keyValueMetadata,
 		Schema:               coalesceSchema(c.Schema, config.Schema),
+		BloomFilters:         coalesceBloomFilters(c.BloomFilters, config.BloomFilters),
 	}
 }
 
@@ -367,6 +374,18 @@ func KeyValueMetadata(key, value string) WriterOption {
 	})
 }
 
+// BloomFilters creates a configuration option which defines the bloom filters
+// that parquet writers should generate.
+//
+// The compute and memory footprint of generating bloom filters for all columns
+// of a parquet schema can be significant, so by default no filters are created
+// and applications need to explicitly declare the columns that they want to
+// create filters for.
+func BloomFilters(filters ...BloomFilterColumn) WriterOption {
+	filters = append([]BloomFilterColumn{}, filters...)
+	return writerOption(func(config *WriterConfig) { config.BloomFilters = filters })
+}
+
 // ColumnBufferSize creates a configuration option which defines the size of
 // row group column buffers.
 //
@@ -455,6 +474,13 @@ func coalesceSortingColumns(s1, s2 []SortingColumn) []SortingColumn {
 	return s2
 }
 
+func coalesceBloomFilters(f1, f2 []BloomFilterColumn) []BloomFilterColumn {
+	if f1 != nil {
+		return f1
+	}
+	return f2
+}
+
 func validatePositiveInt(optionName string, optionValue int) error {
 	if optionValue > 0 {
 		return nil
diff --git a/convert.go b/convert.go
index 5558577..8e909e2 100644
--- a/convert.go
+++ b/convert.go
@@ -437,6 +437,8 @@ func (c *missingColumnChunk) Column() int              { return c.column }
 func (c *missingColumnChunk) Pages() Pages             { return onePage(missingPage{c}) }
 func (c *missingColumnChunk) ColumnIndex() ColumnIndex { return missingColumnIndex{c} }
 func (c *missingColumnChunk) OffsetIndex() OffsetIndex { return missingOffsetIndex{} }
+func (c *missingColumnChunk) BloomFilter() BloomFilter { return missingBloomFilter{} }
+func (c *missingColumnChunk) NumValues() int64         { return 0 }
 
 type missingColumnIndex struct{ *missingColumnChunk }
 
@@ -455,6 +457,12 @@ func (missingOffsetIndex) Offset(int) int64             { return 0 }
 func (missingOffsetIndex) CompressedPageSize(int) int64 { return 0 }
 func (missingOffsetIndex) FirstRowIndex(int) int64      { return 0 }
 
+type missingBloomFilter struct{}
+
+func (missingBloomFilter) ReadAt([]byte, int64) (int, error) { return 0, io.EOF }
+func (missingBloomFilter) Size() int64                       { return 0 }
+func (missingBloomFilter) Check([]byte) (bool, error)        { return false, nil }
+
 type missingPage struct{ *missingColumnChunk }
 
 func (p missingPage) Column() int              { return p.column }
diff --git a/dictionary.go b/dictionary.go
index 5d91502..1ed7d49 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -728,6 +728,14 @@ func (page *indexedPage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *indexedPage) Clone() BufferedPage {
+	return &indexedPage{
+		dict:        page.dict,
+		values:      append([]int32{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *indexedPage) Slice(i, j int64) BufferedPage {
 	return &indexedPage{
 		dict:        page.dict,
@@ -798,6 +806,8 @@ func (col *indexedColumnBuffer) ColumnIndex() ColumnIndex { return indexedColumn
 
 func (col *indexedColumnBuffer) OffsetIndex() OffsetIndex { return indexedOffsetIndex{col} }
 
+func (col *indexedColumnBuffer) BloomFilter() BloomFilter { return nil }
+
 func (col *indexedColumnBuffer) Dictionary() Dictionary { return col.dict }
 
 func (col *indexedColumnBuffer) Pages() Pages { return onePage(col.Page()) }
diff --git a/encoding/delta/binary_packed_decoder.go b/encoding/delta/binary_packed_decoder.go
index 653d41b..6472861 100644
--- a/encoding/delta/binary_packed_decoder.go
+++ b/encoding/delta/binary_packed_decoder.go
@@ -7,7 +7,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -57,10 +56,6 @@ func (d *BinaryPackedDecoder) Reset(r io.Reader) {
 	d.miniBlocks.Reset(d.reader)
 }
 
-func (e *BinaryPackedDecoder) Encoding() format.Encoding {
-	return format.DeltaBinaryPacked
-}
-
 func (d *BinaryPackedDecoder) DecodeInt32(data []int32) (int, error) {
 	decoded := 0
 
diff --git a/encoding/delta/binary_packed_encoder.go b/encoding/delta/binary_packed_encoder.go
index 1d0d0fe..4ced714 100644
--- a/encoding/delta/binary_packed_encoder.go
+++ b/encoding/delta/binary_packed_encoder.go
@@ -5,7 +5,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -46,10 +45,6 @@ func (e *BinaryPackedEncoder) Reset(w io.Writer) {
 	e.miniBlock.Reset(w)
 }
 
-func (e *BinaryPackedEncoder) Encoding() format.Encoding {
-	return format.DeltaBinaryPacked
-}
-
 func (e *BinaryPackedEncoder) EncodeInt32(data []int32) error {
 	firstValue := int32(0)
 	if len(data) > 0 {
diff --git a/encoding/delta/byte_array_decoder.go b/encoding/delta/byte_array_decoder.go
index c22096e..a24f1e7 100644
--- a/encoding/delta/byte_array_decoder.go
+++ b/encoding/delta/byte_array_decoder.go
@@ -6,7 +6,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type ByteArrayDecoder struct {
@@ -36,10 +35,6 @@ func (d *ByteArrayDecoder) Reset(r io.Reader) {
 	d.prefixes = d.prefixes[:0]
 }
 
-func (d *ByteArrayDecoder) Encoding() format.Encoding {
-	return format.DeltaByteArray
-}
-
 func (d *ByteArrayDecoder) DecodeByteArray(data *encoding.ByteArrayList) (int, error) {
 	return d.decode(data.Cap()-data.Len(), func(n int) ([]byte, error) { return data.PushSize(n), nil })
 }
diff --git a/encoding/delta/byte_array_encoder.go b/encoding/delta/byte_array_encoder.go
index ac7207a..653e269 100644
--- a/encoding/delta/byte_array_encoder.go
+++ b/encoding/delta/byte_array_encoder.go
@@ -7,7 +7,6 @@ import (
 	"math"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type ByteArrayEncoder struct {
@@ -31,10 +30,6 @@ func (e *ByteArrayEncoder) Reset(w io.Writer) {
 	e.suffixes.Reset()
 }
 
-func (e *ByteArrayEncoder) Encoding() format.Encoding {
-	return format.DeltaByteArray
-}
-
 func (e *ByteArrayEncoder) EncodeByteArray(data encoding.ByteArrayList) error {
 	return e.encode(data.Len(), data.Index)
 }
diff --git a/encoding/delta/length_byte_array_decoder.go b/encoding/delta/length_byte_array_decoder.go
index ef8b974..ec9b682 100644
--- a/encoding/delta/length_byte_array_decoder.go
+++ b/encoding/delta/length_byte_array_decoder.go
@@ -5,7 +5,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type LengthByteArrayDecoder struct {
@@ -27,10 +26,6 @@ func (d *LengthByteArrayDecoder) Reset(r io.Reader) {
 	d.index = -1
 }
 
-func (d *LengthByteArrayDecoder) Encoding() format.Encoding {
-	return format.DeltaLengthByteArray
-}
-
 func (d *LengthByteArrayDecoder) DecodeByteArray(data *encoding.ByteArrayList) (n int, err error) {
 	if d.index < 0 {
 		if err := d.decodeLengths(); err != nil {
diff --git a/encoding/delta/length_byte_array_encoder.go b/encoding/delta/length_byte_array_encoder.go
index e96ee0e..efbed42 100644
--- a/encoding/delta/length_byte_array_encoder.go
+++ b/encoding/delta/length_byte_array_encoder.go
@@ -6,7 +6,6 @@ import (
 	"math"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type LengthByteArrayEncoder struct {
@@ -25,10 +24,6 @@ func (e *LengthByteArrayEncoder) Reset(w io.Writer) {
 	e.binpack.Reset(w)
 }
 
-func (e *LengthByteArrayEncoder) Encoding() format.Encoding {
-	return format.DeltaLengthByteArray
-}
-
 func (e *LengthByteArrayEncoder) EncodeByteArray(data encoding.ByteArrayList) (err error) {
 	e.lengths = e.lengths[:0]
 
diff --git a/encoding/encoding.go b/encoding/encoding.go
index 44c8ee6..7df9fe7 100644
--- a/encoding/encoding.go
+++ b/encoding/encoding.go
@@ -51,9 +51,6 @@ type Encoder interface {
 	// Calling Reset does not override the bit-width configured on the encoder.
 	Reset(io.Writer)
 
-	// Returns the parquet code for the encoding supported by this encoder.
-	Encoding() format.Encoding
-
 	// Encodes an array of boolean values using this encoder.
 	EncodeBoolean(data []bool) error
 
@@ -117,9 +114,6 @@ type Decoder interface {
 	// Calling Reset does not override the bit-width configured on the decoder.
 	Reset(io.Reader)
 
-	// Returns the parquet code for the encoding supported by this decoder.
-	Encoding() format.Encoding
-
 	// Decodes an array of boolean values using this decoder, returning
 	// the number of decoded values, and io.EOF if the end of the underlying
 	// io.Reader was reached.
diff --git a/encoding/plain/decoder.go b/encoding/plain/decoder.go
index edafb58..fbc4098 100644
--- a/encoding/plain/decoder.go
+++ b/encoding/plain/decoder.go
@@ -8,7 +8,6 @@ import (
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/encoding/rle"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -23,10 +22,6 @@ func NewDecoder(r io.Reader) *Decoder {
 	return &Decoder{reader: r}
 }
 
-func (d *Decoder) Encoding() format.Encoding {
-	return format.Plain
-}
-
 func (d *Decoder) Reset(r io.Reader) {
 	d.reader = r
 
diff --git a/encoding/plain/dictionary.go b/encoding/plain/dictionary.go
index 9d6315a..ce10f9e 100644
--- a/encoding/plain/dictionary.go
+++ b/encoding/plain/dictionary.go
@@ -19,21 +19,13 @@ func (e DictionaryEncoding) CanEncode(t format.Type) bool {
 }
 
 func (e DictionaryEncoding) NewDecoder(r io.Reader) encoding.Decoder {
-	return dictionaryDecoder{NewDecoder(r)}
+	return NewDecoder(r)
 }
 
 func (e DictionaryEncoding) NewEncoder(w io.Writer) encoding.Encoder {
-	return dictionaryEncoder{NewEncoder(w)}
+	return NewEncoder(w)
 }
 
 func (e DictionaryEncoding) String() string {
 	return "DICT_DICTIONARY"
 }
-
-type dictionaryDecoder struct{ *Decoder }
-
-func (d dictionaryDecoder) Encoding() format.Encoding { return format.PlainDictionary }
-
-type dictionaryEncoder struct{ *Encoder }
-
-func (e dictionaryEncoder) Encoding() format.Encoding { return format.PlainDictionary }
diff --git a/encoding/plain/encoder.go b/encoding/plain/encoder.go
index 5478cdf..26d9833 100644
--- a/encoding/plain/encoder.go
+++ b/encoding/plain/encoder.go
@@ -8,7 +8,6 @@ import (
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/encoding/rle"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -23,10 +22,6 @@ func NewEncoder(w io.Writer) *Encoder {
 	return &Encoder{writer: w}
 }
 
-func (e *Encoder) Encoding() format.Encoding {
-	return format.Plain
-}
-
 func (e *Encoder) Reset(w io.Writer) {
 	e.writer = w
 
diff --git a/encoding/rle/decoder.go b/encoding/rle/decoder.go
index 3d505df..b7ebb18 100644
--- a/encoding/rle/decoder.go
+++ b/encoding/rle/decoder.go
@@ -6,7 +6,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -24,10 +23,6 @@ func NewDecoder(r io.Reader) *Decoder {
 	return &Decoder{reader: r}
 }
 
-func (d *Decoder) Encoding() format.Encoding {
-	return format.RLE
-}
-
 func (d *Decoder) BitWidth() int {
 	return int(d.bitWidth)
 }
diff --git a/encoding/rle/dictionary.go b/encoding/rle/dictionary.go
index cfb8968..311e8c1 100644
--- a/encoding/rle/dictionary.go
+++ b/encoding/rle/dictionary.go
@@ -42,10 +42,6 @@ func (d dictionaryDecoder) Reset(r io.Reader) {
 	d.rle.SetBitWidth(0)
 }
 
-func (d dictionaryDecoder) Encoding() format.Encoding {
-	return format.RLEDictionary
-}
-
 func (d dictionaryDecoder) DecodeInt32(data []int32) (int, error) {
 	if d.rle.BitWidth() == 0 {
 		bitWidth, err := d.decodeBitWidth()
@@ -81,10 +77,6 @@ func (e dictionaryEncoder) Reset(w io.Writer) {
 	e.rle.Reset(w)
 }
 
-func (e dictionaryEncoder) Encoding() format.Encoding {
-	return format.RLEDictionary
-}
-
 func (e dictionaryEncoder) EncodeInt32(data []int32) error {
 	bitWidth := bits.MaxLen32(data)
 	if bitWidth == 0 {
diff --git a/encoding/rle/encoder.go b/encoding/rle/encoder.go
index f95b781..e0ff25d 100644
--- a/encoding/rle/encoder.go
+++ b/encoding/rle/encoder.go
@@ -7,7 +7,6 @@ import (
 	"io"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -24,10 +23,6 @@ func NewEncoder(w io.Writer) *Encoder {
 	return &Encoder{writer: w}
 }
 
-func (e *Encoder) Encoding() format.Encoding {
-	return format.RLE
-}
-
 func (e *Encoder) Write(b []byte) (int, error) {
 	return e.writer.Write(b)
 }
diff --git a/file.go b/file.go
index e8d65b1..7afef36 100644
--- a/file.go
+++ b/file.go
@@ -93,6 +93,31 @@ func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 		f.rowGroups[i].init(f, schema, columns, &f.metadata.RowGroups[i])
 	}
 
+	if !c.SkipBloomFilters {
+		h := format.BloomFilterHeader{}
+		p := thrift.CompactProtocol{}
+		s := io.NewSectionReader(r, 0, size)
+		d := thrift.NewDecoder(p.NewReader(s))
+
+		for i := range f.rowGroups {
+			g := &f.rowGroups[i]
+
+			for j := range g.columns {
+				c := &g.columns[j]
+
+				if offset := c.chunk.MetaData.BloomFilterOffset; offset > 0 {
+					s.Seek(offset, io.SeekStart)
+					h = format.BloomFilterHeader{}
+					if err := d.Decode(&h); err != nil {
+						return nil, err
+					}
+					offset, _ = s.Seek(0, io.SeekCurrent)
+					c.bloomFilter = newBloomFilter(r, offset, &h)
+				}
+			}
+		}
+	}
+
 	sortKeyValueMetadata(f.metadata.KeyValueMetadata)
 	return f, nil
 }
@@ -354,6 +379,7 @@ func (s *fileSortingColumn) NullsFirst() bool { return s.nullsFirst }
 type fileColumnChunk struct {
 	file        *File
 	column      *Column
+	bloomFilter *bloomFilter
 	rowGroup    *format.RowGroup
 	columnIndex *format.ColumnIndex
 	offsetIndex *format.OffsetIndex
@@ -407,6 +433,17 @@ func (c *fileColumnChunk) OffsetIndex() OffsetIndex {
 	return (*offsetIndex)(c.offsetIndex)
 }
 
+func (c *fileColumnChunk) BloomFilter() BloomFilter {
+	if c.bloomFilter == nil {
+		return nil
+	}
+	return c.bloomFilter
+}
+
+func (c *fileColumnChunk) NumValues() int64 {
+	return c.chunk.MetaData.NumValues
+}
+
 type filePages struct {
 	column     *fileColumnChunk
 	protocol   thrift.CompactProtocol
@@ -814,14 +851,17 @@ type filePageValueReaderState struct {
 	}
 
 	repetitions struct {
-		decoder encoding.Decoder
+		encoding format.Encoding
+		decoder  encoding.Decoder
 	}
 
 	definitions struct {
-		decoder encoding.Decoder
+		encoding format.Encoding
+		decoder  encoding.Decoder
 	}
 
 	page struct {
+		encoding   format.Encoding
 		decoder    encoding.Decoder
 		compressed *compressedPageReader
 	}
@@ -871,9 +911,17 @@ func (s *filePageValueReaderState) init(columnType Type, column *Column, codec f
 		return fmt.Errorf("cannot read values from page of type %s", h.PageType())
 	}
 
-	s.repetitions.decoder = makeDecoder(s.repetitions.decoder, pageHeader.RepetitionLevelEncoding(), repetitionLevels)
-	s.definitions.decoder = makeDecoder(s.definitions.decoder, pageHeader.DefinitionLevelEncoding(), definitionLevels)
-	s.page.decoder = makeDecoder(s.page.decoder, pageHeader.Encoding(), pageData)
+	repetitionLevelEncoding := pageHeader.RepetitionLevelEncoding()
+	definitionLevelEncoding := pageHeader.DefinitionLevelEncoding()
+	pageEncoding := pageHeader.Encoding()
+
+	s.repetitions.decoder = makeDecoder(s.repetitions.decoder, s.repetitions.encoding, repetitionLevelEncoding, repetitionLevels)
+	s.definitions.decoder = makeDecoder(s.definitions.decoder, s.definitions.encoding, definitionLevelEncoding, definitionLevels)
+	s.page.decoder = makeDecoder(s.page.decoder, s.page.encoding, pageEncoding, pageData)
+
+	s.repetitions.encoding = repetitionLevelEncoding
+	s.definitions.encoding = definitionLevelEncoding
+	s.page.encoding = pageEncoding
 
 	if s.reader == nil {
 		s.reader = newDataPageReader(
@@ -1004,9 +1052,9 @@ func makeCompressedPage(page *compressedPageReader, codec format.CompressionCode
 	return page
 }
 
-func makeDecoder(decoder encoding.Decoder, encoding format.Encoding, input io.Reader) encoding.Decoder {
-	if decoder == nil || encoding != decoder.Encoding() {
-		decoder = LookupEncoding(encoding).NewDecoder(input)
+func makeDecoder(decoder encoding.Decoder, oldEncoding, newEncoding format.Encoding, input io.Reader) encoding.Decoder {
+	if decoder == nil || oldEncoding != newEncoding {
+		decoder = LookupEncoding(newEncoding).NewDecoder(input)
 	} else {
 		decoder.Reset(input)
 	}
diff --git a/page.go b/page.go
index cece882..846b70f 100644
--- a/page.go
+++ b/page.go
@@ -59,6 +59,10 @@ type Page interface {
 type BufferedPage interface {
 	Page
 
+	// Returns a copy of the page which does not share any of the buffers, but
+	// contains the same values, repetition and definition levels.
+	Clone() BufferedPage
+
 	// Returns a new page which is as slice of the receiver between row indexes
 	// i and j.
 	Slice(i, j int64) BufferedPage
@@ -210,6 +214,7 @@ func (page *errorPage) NumRows() int64                 { return 0 }
 func (page *errorPage) NumValues() int64               { return 0 }
 func (page *errorPage) NumNulls() int64                { return 0 }
 func (page *errorPage) Bounds() (min, max Value)       { return }
+func (page *errorPage) Clone() BufferedPage            { return page }
 func (page *errorPage) Slice(i, j int64) BufferedPage  { return page }
 func (page *errorPage) Size() int64                    { return 0 }
 func (page *errorPage) RepetitionLevels() []int8       { return nil }
@@ -288,6 +293,14 @@ func (page *optionalPage) Bounds() (min, max Value) {
 	return page.base.Bounds()
 }
 
+func (page *optionalPage) Clone() BufferedPage {
+	return newOptionalPage(
+		page.base.Clone(),
+		page.maxDefinitionLevel,
+		append([]int8{}, page.definitionLevels...),
+	)
+}
+
 func (page *optionalPage) Slice(i, j int64) BufferedPage {
 	numNulls1 := int64(countLevelsNotEqual(page.definitionLevels[:i], page.maxDefinitionLevel))
 	numNulls2 := int64(countLevelsNotEqual(page.definitionLevels[i:j], page.maxDefinitionLevel))
@@ -408,6 +421,16 @@ func (page *repeatedPage) Bounds() (min, max Value) {
 	return page.base.Bounds()
 }
 
+func (page *repeatedPage) Clone() BufferedPage {
+	return newRepeatedPage(
+		page.base.Clone(),
+		page.maxRepetitionLevel,
+		page.maxDefinitionLevel,
+		append([]int8{}, page.repetitionLevels...),
+		append([]int8{}, page.definitionLevels...),
+	)
+}
+
 func (page *repeatedPage) Slice(i, j int64) BufferedPage {
 	numRows := page.NumRows()
 	if i < 0 || i > numRows {
@@ -588,6 +611,13 @@ func (page *booleanPage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *booleanPage) Clone() BufferedPage {
+	return &booleanPage{
+		values:      append([]bool{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *booleanPage) Slice(i, j int64) BufferedPage {
 	return &booleanPage{
 		values:      page.values[i:j],
@@ -655,6 +685,13 @@ func (page *int32Page) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *int32Page) Clone() BufferedPage {
+	return &int32Page{
+		values:      append([]int32{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *int32Page) Slice(i, j int64) BufferedPage {
 	return &int32Page{
 		values:      page.values[i:j],
@@ -722,6 +759,13 @@ func (page *int64Page) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *int64Page) Clone() BufferedPage {
+	return &int64Page{
+		values:      append([]int64{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *int64Page) Slice(i, j int64) BufferedPage {
 	return &int64Page{
 		values:      page.values[i:j],
@@ -791,6 +835,13 @@ func (page *int96Page) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *int96Page) Clone() BufferedPage {
+	return &int96Page{
+		values:      append([]deprecated.Int96{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *int96Page) Slice(i, j int64) BufferedPage {
 	return &int96Page{
 		values:      page.values[i:j],
@@ -858,6 +909,13 @@ func (page *floatPage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *floatPage) Clone() BufferedPage {
+	return &floatPage{
+		values:      append([]float32{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *floatPage) Slice(i, j int64) BufferedPage {
 	return &floatPage{
 		values:      page.values[i:j],
@@ -925,6 +983,13 @@ func (page *doublePage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *doublePage) Clone() BufferedPage {
+	return &doublePage{
+		values:      append([]float64{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *doublePage) Slice(i, j int64) BufferedPage {
 	return &doublePage{
 		values:      page.values[i:j],
@@ -1030,6 +1095,13 @@ func (page *byteArrayPage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *byteArrayPage) Clone() BufferedPage {
+	return &byteArrayPage{
+		values:      page.values.Clone(),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *byteArrayPage) Slice(i, j int64) BufferedPage {
 	return &byteArrayPage{
 		values:      page.values.Slice(int(i), int(j)),
@@ -1104,6 +1176,14 @@ func (page *fixedLenByteArrayPage) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page *fixedLenByteArrayPage) Clone() BufferedPage {
+	return &fixedLenByteArrayPage{
+		size:        page.size,
+		data:        append([]byte{}, page.data...),
+		columnIndex: page.columnIndex,
+	}
+}
+
 func (page *fixedLenByteArrayPage) Slice(i, j int64) BufferedPage {
 	return &fixedLenByteArrayPage{
 		size:        page.size,
@@ -1168,6 +1248,10 @@ func (page uint32Page) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page uint32Page) Clone() BufferedPage {
+	return uint32Page{page.int32Page.Clone().(*int32Page)}
+}
+
 func (page uint32Page) Slice(i, j int64) BufferedPage {
 	return uint32Page{page.int32Page.Slice(i, j).(*int32Page)}
 }
@@ -1193,6 +1277,10 @@ func (page uint64Page) Bounds() (min, max Value) {
 	return min, max
 }
 
+func (page uint64Page) Clone() BufferedPage {
+	return uint64Page{page.int64Page.Clone().(*int64Page)}
+}
+
 func (page uint64Page) Slice(i, j int64) BufferedPage {
 	return uint64Page{page.int64Page.Slice(i, j).(*int64Page)}
 }
diff --git a/page_reader.go b/page_reader.go
index 2282994..0e2425c 100644
--- a/page_reader.go
+++ b/page_reader.go
@@ -117,7 +117,7 @@ func (r *dataPageReader) ReadValues(values []Value) (int, error) {
 				// decoded levels.
 				err = fmt.Errorf("after reading %d/%d values: %w", r.numValues-r.remain, r.numValues, io.ErrUnexpectedEOF)
 			}
-			return read, fmt.Errorf("decoding %s values from data page of column %d: %w", r.values.Decoder().Encoding(), ^r.columnIndex, err)
+			return read, fmt.Errorf("decoding values from data page of column %d: %w", ^r.columnIndex, err)
 		}
 
 		for i, j := n-1, len(definitionLevels)-1; j >= 0; j-- {
diff --git a/row_group.go b/row_group.go
index 09efaf6..5424383 100644
--- a/row_group.go
+++ b/row_group.go
@@ -15,10 +15,10 @@ type RowGroup interface {
 	// Returns the number of rows in the group.
 	NumRows() int64
 
-	// Returns the number of columns in the group.
+	// Returns the number of leaf columns in the group.
 	NumColumns() int
 
-	// Returns the column at the given index in the group.
+	// Returns the leaf column at the given index in the group.
 	Column(int) ColumnChunk
 
 	// Returns the schema of rows in the group.
@@ -375,6 +375,14 @@ func (c *seekColumnChunk) OffsetIndex() OffsetIndex {
 	return c.base.OffsetIndex()
 }
 
+func (c *seekColumnChunk) BloomFilter() BloomFilter {
+	return c.base.BloomFilter()
+}
+
+func (c *seekColumnChunk) NumValues() int64 {
+	return c.base.NumValues()
+}
+
 type emptyRowGroup struct {
 	schema  *Schema
 	columns []emptyColumnChunk
@@ -409,6 +417,14 @@ func (c *emptyColumnChunk) Column() int              { return c.column }
 func (c *emptyColumnChunk) Pages() Pages             { return emptyPages{} }
 func (c *emptyColumnChunk) ColumnIndex() ColumnIndex { return &emptyColumnIndex }
 func (c *emptyColumnChunk) OffsetIndex() OffsetIndex { return &emptyOffsetIndex }
+func (c *emptyColumnChunk) BloomFilter() BloomFilter { return emptyBloomFilter{} }
+func (c *emptyColumnChunk) NumValues() int64         { return 0 }
+
+type emptyBloomFilter struct{}
+
+func (emptyBloomFilter) ReadAt([]byte, int64) (int, error) { return 0, io.EOF }
+func (emptyBloomFilter) Size() int64                       { return 0 }
+func (emptyBloomFilter) Check([]byte) (bool, error)        { return false, nil }
 
 type emptyRowReader struct{ schema *Schema }
 
diff --git a/writer.go b/writer.go
index 2aa8e0c..1cea895 100644
--- a/writer.go
+++ b/writer.go
@@ -165,6 +165,7 @@ func (w *Writer) WriteRowGroup(rowGroup RowGroup) (int64, error) {
 	if err := w.writer.flush(); err != nil {
 		return 0, err
 	}
+	w.writer.configureBloomFilters(rowGroup)
 	n, err := CopyRows(w.writer, rowGroup.Rows())
 	if err != nil {
 		return n, err
@@ -282,6 +283,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 			columnPath:         leaf.path,
 			columnType:         columnType,
 			columnIndex:        columnType.NewColumnIndexer(config.ColumnIndexSizeLimit),
+			columnFilter:       searchBloomFilterColumn(config.BloomFilters, leaf.path),
 			compression:        compression,
 			dictionary:         dictionary,
 			dataPageType:       dataPageType,
@@ -324,8 +326,9 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 			c.encodings = addEncoding(c.encodings, format.Plain)
 		}
 
-		c.encodings = addEncoding(c.encodings, encoding.Encoding())
 		c.page.encoder = encoding.NewEncoder(nil)
+		c.page.encoding = encoding.Encoding()
+		c.encodings = addEncoding(c.encodings, c.page.encoding)
 		sortPageEncodings(c.encodings)
 
 		w.columns = append(w.columns, c)
@@ -406,6 +409,14 @@ func (w *writer) writeFileHeader() error {
 	return nil
 }
 
+func (w *writer) configureBloomFilters(rowGroup RowGroup) {
+	for i, c := range w.columns {
+		if c.columnFilter != nil {
+			c.page.filter = c.newBloomFilterEncoder(rowGroup.Column(i).NumValues())
+		}
+	}
+}
+
 func (w *writer) writeFileFooter() error {
 	// The page index is composed of two sections: column and offset indexes.
 	// They are written after the row groups, right before the footer (which
@@ -492,6 +503,9 @@ func (w *writer) writeRowGroup(rowGroupSchema *Schema, rowGroupSortingColumns []
 		if err := c.flush(); err != nil {
 			return 0, err
 		}
+		if err := c.flushFilterPages(); err != nil {
+			return 0, err
+		}
 	}
 
 	if err := w.writeFileHeader(); err != nil {
@@ -499,6 +513,15 @@ func (w *writer) writeRowGroup(rowGroupSchema *Schema, rowGroupSortingColumns []
 	}
 	fileOffset := w.writer.offset
 
+	for _, c := range w.columns {
+		if c.page.filter != nil {
+			c.columnChunk.MetaData.BloomFilterOffset = w.writer.offset
+			if err := c.writeBloomFilter(&w.writer); err != nil {
+				return 0, err
+			}
+		}
+	}
+
 	for i, c := range w.columns {
 		w.columnIndex[i] = format.ColumnIndex(c.columnIndex.ColumnIndex())
 
@@ -603,6 +626,7 @@ type writerColumn struct {
 	insert func(*writerColumn, Row) error
 	commit func(*writerColumn) error
 	values []Value
+	filter []BufferedPage
 
 	pool  PageBufferPool
 	pages []io.ReadWriter
@@ -611,6 +635,7 @@ type writerColumn struct {
 	columnType   Type
 	columnIndex  ColumnIndexer
 	columnBuffer ColumnBuffer
+	columnFilter BloomFilterColumn
 	compression  compress.Codec
 	dictionary   Dictionary
 
@@ -639,8 +664,10 @@ type writerColumn struct {
 
 	page struct {
 		buffer       *bytes.Buffer
+		filter       *bloomFilterEncoder
 		compressed   compress.Writer
 		uncompressed offsetTrackingWriter
+		encoding     format.Encoding
 		encoder      encoding.Encoder
 	}
 
@@ -674,9 +701,13 @@ func (c *writerColumn) reset() {
 	for _, page := range c.pages {
 		c.pool.PutPageBuffer(page)
 	}
+	for i := range c.filter {
+		c.filter[i] = nil
+	}
 	for i := range c.pages {
 		c.pages[i] = nil
 	}
+	c.filter = c.filter[:0]
 	c.pages = c.pages[:0]
 	c.numRows = 0
 	c.numValues = 0
@@ -694,6 +725,11 @@ func (c *writerColumn) reset() {
 	// the number of pages should be roughly the same between row groups written
 	// by the writer.
 	c.offsetIndex.PageLocations = make([]format.PageLocation, 0, cap(c.offsetIndex.PageLocations))
+	// Bloom filters may change in size between row groups; we may want to
+	// optimize this by retaining the filter and reusing it if needed, but
+	// for now we take the simpler approach of freeing it and having the
+	// write path lazily reallocate it if the writer is reused.
+	c.page.filter = nil
 }
 
 func (c *writerColumn) totalRowCount() int64 {
@@ -717,6 +753,24 @@ func (c *writerColumn) flush() (err error) {
 	return err
 }
 
+func (c *writerColumn) flushFilterPages() error {
+	if c.columnFilter != nil {
+		numValues := int64(0)
+		for _, page := range c.filter {
+			numValues += page.NumValues()
+		}
+		if c.page.filter == nil {
+			c.page.filter = c.newBloomFilterEncoder(numValues)
+		}
+		for _, page := range c.filter {
+			if err := page.WriteTo(c.page.filter); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func (c *writerColumn) insertRepeated(row Row) error {
 	c.values = append(c.values, row...)
 	return nil
@@ -741,6 +795,14 @@ func (c *writerColumn) newColumnBuffer() ColumnBuffer {
 	return column
 }
 
+func (c *writerColumn) newBloomFilterEncoder(numRows int64) *bloomFilterEncoder {
+	const bitsPerValue = 10 // TODO: make this configurable
+	return newBloomFilterEncoder(
+		c.columnFilter.NewFilter(numRows, bitsPerValue),
+		c.columnFilter.Hash(),
+	)
+}
+
 func (c *writerColumn) WriteRow(row Row) error {
 	if c.columnBuffer == nil {
 		// Lazily create the row group column so we don't need to allocate it if
@@ -834,6 +896,18 @@ func (c *writerColumn) writePageValues(page ValueReader) (numValues int64, err e
 	return numValues, err
 }
 
+func (c *writerColumn) writeBloomFilter(w io.Writer) error {
+	e := thrift.NewEncoder(c.header.protocol.NewWriter(w))
+	h := bloomFilterHeader(c.columnFilter)
+	b := c.page.filter.Bytes()
+	h.NumBytes = int32(len(b))
+	if err := e.Encode(&h); err != nil {
+		return err
+	}
+	_, err := w.Write(b)
+	return err
+}
+
 func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 	numValues := page.NumValues()
 	if numValues == 0 {
@@ -888,6 +962,15 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 		}
 	}
 
+	switch {
+	case c.page.filter != nil:
+		if err := page.WriteTo(c.page.filter); err != nil {
+			return 0, err
+		}
+	case c.columnFilter != nil:
+		c.filter = append(c.filter, page.Clone())
+	}
+
 	statistics := format.Statistics{}
 	if c.writePageStats {
 		statistics = c.makePageStatistics(page)
@@ -907,7 +990,6 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 	levelsByteLength := repetitionLevelsByteLength + definitionLevelsByteLength
 	uncompressedPageSize := c.page.uncompressed.offset + int64(levelsByteLength)
 	compressedPageSize := c.page.buffer.Len()
-	encoding := c.page.encoder.Encoding()
 
 	pageHeader := &format.PageHeader{
 		Type:                 c.dataPageType,
@@ -922,7 +1004,7 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 	case format.DataPage:
 		pageHeader.DataPageHeader = &format.DataPageHeader{
 			NumValues:               int32(numValues),
-			Encoding:                encoding,
+			Encoding:                c.page.encoding,
 			DefinitionLevelEncoding: format.RLE,
 			RepetitionLevelEncoding: format.RLE,
 			Statistics:              statistics,
@@ -932,7 +1014,7 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 			NumValues:                  int32(numValues),
 			NumNulls:                   int32(numNulls),
 			NumRows:                    int32(numRows),
-			Encoding:                   encoding,
+			Encoding:                   c.page.encoding,
 			DefinitionLevelsByteLength: definitionLevelsByteLength,
 			RepetitionLevelsByteLength: repetitionLevelsByteLength,
 			IsCompressed:               &c.isCompressed,
@@ -953,6 +1035,27 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 }
 
 func (c *writerColumn) writeCompressedPage(page CompressedPage) (int64, error) {
+	switch {
+	case c.page.filter != nil:
+		// TODO: modify the Buffer method to accept some kind of buffer pool as
+		// argument so we can use a pre-allocated page buffer to load the page
+		// and reduce the memory footprint.
+		bufferedPage := page.Buffer()
+		// The compressed page must be decompressed here in order to generate
+		// the bloom filter. Note that we don't re-compress it which still saves
+		// most of the compute cost (compression algorithms are usually designed
+		// to make decompressing much cheaper than compressing since it happens
+		// more often).
+		if err := bufferedPage.WriteTo(c.page.filter); err != nil {
+			return 0, err
+		}
+	case c.columnFilter != nil:
+		// When a column filter is configured but no page filter was allocated,
+		// we need to buffer the page in order to have access to the number of
+		// values and properly size the bloom filter when writing the row group.
+		c.filter = append(c.filter, page.Buffer())
+	}
+
 	pageHeader := &format.PageHeader{
 		UncompressedPageSize: int32(page.Size()),
 		CompressedPageSize:   int32(page.PageSize()),


Test patch:
diff --git a/bloom/block_test.go b/bloom/block_test.go
new file mode 100644
index 0000000..b1d5277
--- /dev/null
+++ b/bloom/block_test.go
@@ -0,0 +1,45 @@
+package bloom_test
+
+import (
+	"math"
+	"testing"
+
+	"github.com/segmentio/parquet-go/bloom"
+)
+
+func TestBlock(t *testing.T) {
+	for i := uint64(0); i < math.MaxUint32; i = (i * 2) + 1 {
+		x := uint32(i)
+		b := bloom.Block{}
+		b.Insert(x)
+		if !b.Check(x) {
+			t.Fatalf("bloom filter block does not contain the value that was inserted: %d", x)
+		}
+		if b.Check(x - 1) {
+			t.Fatalf("bloom filter block contains value that was not inserted: %d", ^x)
+		}
+		if b.Check(x + 1) {
+			t.Fatalf("bloom filter block contains value that was not inserted: %d", ^x)
+		}
+		if b.Check(^x) {
+			t.Fatalf("bloom filter block contains value that was not inserted: %d", ^x)
+		}
+	}
+}
+
+func BenchmarkBlockInsert(b *testing.B) {
+	x := bloom.Block{}
+	for i := 0; i < b.N; i++ {
+		x.Insert(uint32(i))
+	}
+	b.SetBytes(bloom.BlockSize)
+}
+
+func BenchmarkBlockCheck(b *testing.B) {
+	x := bloom.Block{}
+	x.Insert(42)
+	for i := 0; i < b.N; i++ {
+		x.Check(42)
+	}
+	b.SetBytes(bloom.BlockSize)
+}
diff --git a/bloom/bloom_test.go b/bloom/bloom_test.go
new file mode 100644
index 0000000..ff186fa
--- /dev/null
+++ b/bloom/bloom_test.go
@@ -0,0 +1,37 @@
+package bloom
+
+import (
+	"math/rand"
+	"testing"
+)
+
+// Test file for internal functions of the bloom package.
+var global4x64 [4]uint64
+
+func TestFasthash(t *testing.T) {
+	r := rand.NewSource(0).(rand.Source64)
+
+	src := [4]uint64{r.Uint64(), r.Uint64(), r.Uint64(), r.Uint64()}
+	dst := [4]uint64{0, 0, 0, 0}
+	exp := [4]uint64{483, 125, 335, 539}
+	mod := int32(1024)
+
+	fasthash4x64(&dst, &src, mod)
+
+	if dst != exp {
+		t.Errorf("got=%v want=%v", dst, exp)
+	}
+}
+
+func BenchmarkFasthash(b *testing.B) {
+	src := [4]uint64{}
+	dst := [4]uint64{}
+	mod := int32(1024)
+
+	for i := 0; i < b.N; i++ {
+		fasthash4x64(&dst, &src, mod)
+	}
+
+	b.SetBytes(32)
+	global4x64 = dst // use it so the loop isn't optimized away
+}
diff --git a/bloom/filter_test.go b/bloom/filter_test.go
new file mode 100644
index 0000000..161efce
--- /dev/null
+++ b/bloom/filter_test.go
@@ -0,0 +1,142 @@
+package bloom_test
+
+import (
+	"bytes"
+	"math/rand"
+	"testing"
+
+	"github.com/segmentio/parquet-go/bloom"
+)
+
+func TestSplitBlockFilter(t *testing.T) {
+	const N = 1000
+	const S = 3
+	f := make(bloom.SplitBlockFilter, bloom.NumSplitBlocksOf(N, 10))
+	p := rand.New(rand.NewSource(S))
+
+	// Half of the values are inserted individually.
+	for i := 0; i < N/2; i++ {
+		f.Insert(p.Uint64())
+	}
+	// The other half is inserted as a bulk operation.
+	b := make([]uint64, N/2)
+	for i := range b {
+		b[i] = p.Uint64()
+	}
+	f.InsertBulk(b)
+
+	if f.Block(0) == nil {
+		t.Fatal("looking up filter block returned impossible nil value")
+	}
+
+	for _, test := range []struct {
+		scenario string
+		filter   bloom.Filter
+	}{
+		{scenario: "filter", filter: f},
+		{scenario: "reader", filter: newSerializedFilter(f.Bytes())},
+	} {
+		t.Run(test.scenario, func(t *testing.T) {
+			p.Seed(S)
+			falsePositives := 0
+
+			for i := 0; i < N; i++ {
+				x := p.Uint64()
+
+				if !test.filter.Check(x) {
+					t.Fatalf("bloom filter block does not contain the value #%d that was inserted: %d", i, x)
+				}
+				if test.filter.Check(^x) {
+					falsePositives++
+				}
+			}
+
+			if r := (float64(falsePositives) / N); r > 0.01 {
+				t.Fatalf("bloom filter triggered too many false positives: %g%%", r*100)
+			}
+		})
+	}
+
+	t.Run("Reset", func(t *testing.T) {
+		allZeros := true
+		for _, b := range f.Bytes() {
+			if b != 0 {
+				allZeros = false
+				break
+			}
+		}
+		if allZeros {
+			t.Fatal("bloom filter bytes were all zero after inserting keys")
+		}
+		f.Reset()
+		for i, b := range f.Bytes() {
+			if b != 0 {
+				t.Fatalf("bloom filter byte at index %d was not zero after resetting the filter: %02X", i, b)
+			}
+		}
+	})
+}
+
+func TestSplitBlockFilterBug1(t *testing.T) {
+	// This test exercises the case where we bulk insert a single key in the
+	// filter, which skips the core of the optimized assembly routines and runs
+	// through the loop handling tails of remaining keys after consuming groups
+	// of two or more.
+	//
+	// The use of quick.Check in bloom filter tests of the parquet package had
+	// uncovered a bug which was reproduced here in isolation when debugging.
+	h := [1]uint64{0b1000101001000001001001111000000100011011001000011110011100110000}
+	f := make(bloom.SplitBlockFilter, 1)
+	f.InsertBulk(h[:])
+	if !f.Check(h[0]) {
+		t.Error("value inserted in the filter was not found")
+	}
+}
+
+type serializedFilter struct {
+	bytes.Reader
+}
+
+func (f *serializedFilter) Check(x uint64) bool {
+	ok, _ := bloom.CheckSplitBlock(&f.Reader, f.Size(), x)
+	return ok
+}
+
+func newSerializedFilter(b []byte) *serializedFilter {
+	f := new(serializedFilter)
+	f.Reset(b)
+	return f
+}
+
+func BenchmarkFilterInsertBulk(b *testing.B) {
+	f := make(bloom.SplitBlockFilter, 99)
+	x := make([]uint64, 16)
+	r := rand.NewSource(0).(rand.Source64)
+
+	for i := range x {
+		x[i] = r.Uint64()
+	}
+
+	for i := 0; i < b.N; i++ {
+		f.InsertBulk(x)
+	}
+
+	b.SetBytes(bloom.BlockSize * int64(len(x)))
+}
+
+func BenchmarkFilterInsert(b *testing.B) {
+	f := make(bloom.SplitBlockFilter, 1)
+	for i := 0; i < b.N; i++ {
+		f.Insert(uint64(i))
+	}
+	b.SetBytes(bloom.BlockSize)
+}
+
+func BenchmarkFilterCheck(b *testing.B) {
+	f := make(bloom.SplitBlockFilter, 1)
+	f.Insert(42)
+	for i := 0; i < b.N; i++ {
+		f.Check(42)
+	}
+	b.SetBytes(bloom.BlockSize)
+}
diff --git a/bloom/xxhash/sum64uint_test.go b/bloom/xxhash/sum64uint_test.go
new file mode 100644
index 0000000..2fa9b69
--- /dev/null
+++ b/bloom/xxhash/sum64uint_test.go
@@ -0,0 +1,284 @@
+package xxhash_test
+
+import (
+	"encoding/binary"
+	"testing"
+	"testing/quick"
+	"time"
+
+	"github.com/segmentio/parquet-go/bloom/xxhash"
+)
+
+func TestSumUint8(t *testing.T) {
+	b := [1]byte{0: 42}
+	h := xxhash.Sum64Uint8(42)
+	x := xxhash.Sum64(b[:])
+	if h != x {
+		t.Errorf("got %064b; want %064b", h, x)
+	}
+}
+
+func TestSumUint16(t *testing.T) {
+	b := [2]byte{0: 42}
+	h := xxhash.Sum64Uint16(42)
+	x := xxhash.Sum64(b[:])
+	if h != x {
+		t.Errorf("got %064b; want %064b", h, x)
+	}
+}
+
+func TestSumUint32(t *testing.T) {
+	b := [4]byte{0: 42}
+	h := xxhash.Sum64Uint32(42)
+	x := xxhash.Sum64(b[:])
+	if h != x {
+		t.Errorf("got %064b; want %064b", h, x)
+	}
+}
+
+func TestSumUint64(t *testing.T) {
+	b := [8]byte{0: 42}
+	h := xxhash.Sum64Uint64(42)
+	x := xxhash.Sum64(b[:])
+	if h != x {
+		t.Errorf("got %064b; want %064b", h, x)
+	}
+}
+
+func TestSumUint128(t *testing.T) {
+	b := [16]byte{0: 42}
+	h := xxhash.Sum64Uint128(b)
+	x := xxhash.Sum64(b[:])
+	if h != x {
+		t.Errorf("got %064b; want %064b", h, x)
+	}
+}
+
+func TestMultiSum64Uint8(t *testing.T) {
+	f := func(v []uint8) bool {
+		h := make([]uint64, len(v))
+		n := xxhash.MultiSum64Uint8(h, v)
+		if n != len(v) {
+			t.Errorf("return value mismatch: got %d; want %d", n, len(v))
+			return false
+		}
+		for i := range h {
+			x := xxhash.Sum64(v[i : i+1])
+			if h[i] != x {
+				t.Errorf("sum at index %d mismatch: got %064b; want %064b", i, h[i], x)
+				return false
+			}
+		}
+		return true
+	}
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestMultiSum64Uint16(t *testing.T) {
+	f := func(v []uint16) bool {
+		h := make([]uint64, len(v))
+		n := xxhash.MultiSum64Uint16(h, v)
+		if n != len(v) {
+			t.Errorf("return value mismatch: got %d; want %d", n, len(v))
+			return false
+		}
+		for i := range h {
+			b := [2]byte{}
+			binary.LittleEndian.PutUint16(b[:], v[i])
+			x := xxhash.Sum64(b[:])
+			if h[i] != x {
+				t.Errorf("sum at index %d mismatch: got %064b; want %064b", i, h[i], x)
+				return false
+			}
+		}
+		return true
+	}
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestMultiSum64Uint32(t *testing.T) {
+	f := func(v []uint32) bool {
+		h := make([]uint64, len(v))
+		n := xxhash.MultiSum64Uint32(h, v)
+		if n != len(v) {
+			t.Errorf("return value mismatch: got %d; want %d", n, len(v))
+			return false
+		}
+		for i := range h {
+			b := [4]byte{}
+			binary.LittleEndian.PutUint32(b[:], v[i])
+			x := xxhash.Sum64(b[:])
+			if h[i] != x {
+				t.Errorf("sum at index %d mismatch: got %064b; want %064b", i, h[i], x)
+				return false
+			}
+		}
+		return true
+	}
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestMultiSum64Uint64(t *testing.T) {
+	f := func(v []uint64) bool {
+		h := make([]uint64, len(v))
+		n := xxhash.MultiSum64Uint64(h, v)
+		if n != len(v) {
+			t.Errorf("return value mismatch: got %d; want %d", n, len(v))
+			return false
+		}
+		for i := range h {
+			b := [8]byte{}
+			binary.LittleEndian.PutUint64(b[:], v[i])
+			x := xxhash.Sum64(b[:])
+			if h[i] != x {
+				t.Errorf("sum at index %d mismatch: got %064b; want %064b", i, h[i], x)
+				return false
+			}
+		}
+		return true
+	}
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestMultiSum64Uint128(t *testing.T) {
+	f := func(v [][16]byte) bool {
+		h := make([]uint64, len(v))
+		n := xxhash.MultiSum64Uint128(h, v)
+		if n != len(v) {
+			t.Errorf("return value mismatch: got %d; want %d", n, len(v))
+			return false
+		}
+		for i := range h {
+			x := xxhash.Sum64(v[i][:])
+			if h[i] != x {
+				t.Errorf("sum at index %d mismatch: got %064b; want %064b", i, h[i], x)
+				return false
+			}
+		}
+		return true
+	}
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
+
+func reportThroughput(b *testing.B, loops, count int, start time.Time) {
+	throughput := float64(loops*count) / time.Since(start).Seconds()
+	// Measure the throughput of writes to the output buffer;
+	// it makes the results comparable across benchmarks that
+	// have inputs of different sizes.
+	b.SetBytes(8 * int64(count))
+	b.ReportMetric(0, "ns/op")
+	b.ReportMetric(throughput, "hash/s")
+}
+
+func BenchmarkMultiSum64Uint8(b *testing.B) {
+	for _, bb := range benchmarks {
+		if bb.n < 16 {
+			// Skip smaller inputs that are not relevant to benchmarks that
+			// exercise bulk operations.
+			continue
+		}
+		in := make([]uint8, bb.n)
+		for i := range in {
+			in[i] = uint8(i)
+		}
+		out := make([]uint64, len(in))
+		b.Run(bb.name, func(b *testing.B) {
+			start := time.Now()
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.MultiSum64Uint8(out, in)
+			}
+			reportThroughput(b, b.N, len(out), start)
+		})
+	}
+}
+
+func BenchmarkMultiSum64Uint16(b *testing.B) {
+	for _, bb := range benchmarks {
+		if bb.n < 16 {
+			continue
+		}
+		in := make([]uint16, bb.n)
+		for i := range in {
+			in[i] = uint16(i)
+		}
+		out := make([]uint64, len(in))
+		b.Run(bb.name, func(b *testing.B) {
+			start := time.Now()
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.MultiSum64Uint16(out, in)
+			}
+			reportThroughput(b, b.N, len(out), start)
+		})
+	}
+}
+
+func BenchmarkMultiSum64Uint32(b *testing.B) {
+	for _, bb := range benchmarks {
+		if bb.n < 4 {
+			continue
+		}
+		in := make([]uint32, bb.n/4)
+		for i := range in {
+			in[i] = uint32(i)
+		}
+		out := make([]uint64, len(in))
+		b.Run(bb.name, func(b *testing.B) {
+			start := time.Now()
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.MultiSum64Uint32(out, in)
+			}
+			reportThroughput(b, b.N, len(out), start)
+		})
+	}
+}
+
+func BenchmarkMultiSum64Uint64(b *testing.B) {
+	for _, bb := range benchmarks {
+		if bb.n < 16 {
+			continue
+		}
+		in := make([]uint64, bb.n/8)
+		for i := range in {
+			in[i] = uint64(i)
+		}
+		out := make([]uint64, len(in))
+		b.Run(bb.name, func(b *testing.B) {
+			start := time.Now()
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.MultiSum64Uint64(out, in)
+			}
+			reportThroughput(b, b.N, len(out), start)
+		})
+	}
+}
+
+func BenchmarkMultiSum64Uint128(b *testing.B) {
+	for _, bb := range benchmarks {
+		if bb.n < 16 {
+			continue
+		}
+		in := make([][16]byte, bb.n/16)
+		for i := range in {
+			binary.LittleEndian.PutUint64(in[i][:8], uint64(i))
+			binary.LittleEndian.PutUint64(in[i][8:], uint64(i))
+		}
+		out := make([]uint64, len(in))
+		b.Run(bb.name, func(b *testing.B) {
+			start := time.Now()
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.MultiSum64Uint128(out, in)
+			}
+			reportThroughput(b, b.N, len(out), start)
+		})
+	}
+}
diff --git a/bloom/xxhash/xxhash_test.go b/bloom/xxhash/xxhash_test.go
new file mode 100644
index 0000000..35ba9bb
--- /dev/null
+++ b/bloom/xxhash/xxhash_test.go
@@ -0,0 +1,59 @@
+package xxhash_test
+
+import (
+	"testing"
+
+	"github.com/segmentio/parquet-go/bloom/xxhash"
+)
+
+func TestSum64(t *testing.T) {
+	for _, tt := range []struct {
+		name  string
+		input string
+		want  uint64
+	}{
+		{"empty", "", 0xef46db3751d8e999},
+		{"a", "a", 0xd24ec4f1a98c6e5b},
+		{"as", "as", 0x1c330fb2d66be179},
+		{"asd", "asd", 0x631c37ce72a97393},
+		{"asdf", "asdf", 0x415872f599cea71e},
+		{
+			"len=63",
+			// Exactly 63 characters, which exercises all code paths.
+			"Call me Ishmael. Some years ago--never mind how long precisely-",
+			0x02a2e85470d6fd96,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := xxhash.Sum64([]byte(tt.input)); got != tt.want {
+				t.Fatalf("Sum64: got 0x%x; want 0x%x", got, tt.want)
+			}
+		})
+	}
+}
+
+var benchmarks = []struct {
+	name string
+	n    int64
+}{
+	{"4B", 4},
+	{"16B", 16},
+	{"100B", 100},
+	{"4KB", 4e3},
+	{"10MB", 10e6},
+}
+
+func BenchmarkSum64(b *testing.B) {
+	for _, bb := range benchmarks {
+		in := make([]byte, bb.n)
+		for i := range in {
+			in[i] = byte(i)
+		}
+		b.Run(bb.name, func(b *testing.B) {
+			b.SetBytes(bb.n)
+			for i := 0; i < b.N; i++ {
+				_ = xxhash.Sum64(in)
+			}
+		})
+	}
+}
diff --git a/bloom_test.go b/bloom_test.go
new file mode 100644
index 0000000..6fd803a
--- /dev/null
+++ b/bloom_test.go
@@ -0,0 +1,250 @@
+package parquet
+
+import (
+	"encoding/binary"
+	"math"
+	"math/rand"
+	"testing"
+	"testing/quick"
+
+	"github.com/segmentio/parquet-go/bloom"
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+func TestBloomFilterEncoder(t *testing.T) {
+	newFilter := func(numValues int) *bloomFilterEncoder {
+		return newBloomFilterEncoder(
+			make(bloom.SplitBlockFilter, bloom.NumSplitBlocksOf(int64(numValues), 11)),
+			bloom.XXH64{},
+		)
+	}
+
+	check := func(e *bloomFilterEncoder, b []byte) bool {
+		return e.filter.Check(e.hash.Sum64(b))
+	}
+
+	tests := []struct {
+		scenario string
+		function interface{}
+	}{
+		{
+			scenario: "BOOLEAN",
+			function: func(values []bool) bool {
+				f := newFilter(len(values))
+				f.EncodeBoolean(values)
+				for _, v := range values {
+					b := make([]byte, 1)
+					if v {
+						b[0] = 1
+					}
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "INT8",
+			function: func(values []int8) bool {
+				f := newFilter(len(values))
+				f.EncodeInt8(values)
+				for _, v := range values {
+					b := make([]byte, 1)
+					b[0] = byte(v)
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "INT16",
+			function: func(values []int16) bool {
+				f := newFilter(len(values))
+				f.EncodeInt16(values)
+				for _, v := range values {
+					b := make([]byte, 2)
+					binary.LittleEndian.PutUint16(b, uint16(v))
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "INT32",
+			function: func(values []int32) bool {
+				f := newFilter(len(values))
+				f.EncodeInt32(values)
+				for _, v := range values {
+					b := make([]byte, 4)
+					binary.LittleEndian.PutUint32(b, uint32(v))
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "INT64",
+			function: func(values []int64) bool {
+				f := newFilter(len(values))
+				f.EncodeInt64(values)
+				for _, v := range values {
+					b := make([]byte, 8)
+					binary.LittleEndian.PutUint64(b, uint64(v))
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "INT96",
+			function: func(values []deprecated.Int96) bool {
+				f := newFilter(len(values))
+				f.EncodeInt96(values)
+				for _, v := range values {
+					b := make([]byte, 12)
+					binary.LittleEndian.PutUint32(b[0:4], v[0])
+					binary.LittleEndian.PutUint32(b[4:8], v[1])
+					binary.LittleEndian.PutUint32(b[8:12], v[2])
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "FLOAT",
+			function: func(values []float32) bool {
+				f := newFilter(len(values))
+				f.EncodeFloat(values)
+				for _, v := range values {
+					b := make([]byte, 4)
+					binary.LittleEndian.PutUint32(b, math.Float32bits(v))
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "DOUBLE",
+			function: func(values []float64) bool {
+				f := newFilter(len(values))
+				f.EncodeDouble(values)
+				for _, v := range values {
+					b := make([]byte, 8)
+					binary.LittleEndian.PutUint64(b, math.Float64bits(v))
+					if !check(f, b) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "BYTE_ARRAY",
+			function: func(values [][]byte) bool {
+				a := encoding.ByteArrayList{}
+				for _, v := range values {
+					a.Push(v)
+				}
+				f := newFilter(len(values))
+				f.EncodeByteArray(a)
+				for _, v := range values {
+					if !f.Check(v) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+
+		{
+			scenario: "FIXED_LEN_BYTE_ARRAY",
+			function: func(values [][]byte) bool {
+				a := encoding.ByteArrayList{}
+				for _, v := range values {
+					a.Push(v)
+				}
+				f := newFilter(len(values))
+				f.EncodeByteArray(a)
+				for _, v := range values {
+					if !f.Check(v) {
+						return false
+					}
+				}
+				return true
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.scenario, func(t *testing.T) {
+			if err := quick.Check(test.function, nil); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+
+	t.Run("Reset", func(t *testing.T) {
+		f := newFilter(1)
+		f.EncodeBoolean([]bool{false, true})
+
+		allZeros := true
+		for _, b := range f.Bytes() {
+			if b != 0 {
+				allZeros = false
+				break
+			}
+		}
+		if allZeros {
+			t.Fatal("bloom filter bytes were all zero after encoding values")
+		}
+
+		f.Reset(nil)
+		for i, b := range f.Bytes() {
+			if b != 0 {
+				t.Fatalf("bloom filter byte at index %d was not zero after resetting the encoder: %02X", i, b)
+			}
+		}
+	})
+}
+
+func BenchmarkBloomFilterEncoder(b *testing.B) {
+	const N = 1000
+	f := newBloomFilterEncoder(
+		make(bloom.SplitBlockFilter, bloom.NumSplitBlocksOf(N, 10)),
+		bloom.XXH64{},
+	)
+
+	v := make([]int64, N)
+	r := rand.NewSource(10)
+	for i := range v {
+		v[i] = r.Int63()
+	}
+
+	for i := 0; i < b.N; i++ {
+		f.EncodeInt64(v)
+	}
+
+	b.SetBytes(8 * N)
+}
diff --git a/buffer_test.go b/buffer_test.go
index 1fb8211..f1f0077 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -2,10 +2,12 @@ package parquet_test
 
 import (
 	"bytes"
+	"encoding/binary"
 	"io"
 	"math"
 	"sort"
 	"testing"
+	"testing/quick"
 
 	"github.com/segmentio/parquet-go"
 	"github.com/segmentio/parquet-go/encoding"
@@ -339,3 +341,86 @@ func testBuffer(t *testing.T, node parquet.Node, reader parquet.ValueReader, buf
 		}
 	}
 }
+
+func TestBufferGenerateBloomFilters(t *testing.T) {
+	type Point3D struct {
+		X float64
+		Y float64
+		Z float64
+	}
+
+	f := func(rows []Point3D) bool {
+		if len(rows) == 0 { // TODO: support writing files with no rows
+			return true
+		}
+
+		output := new(bytes.Buffer)
+		buffer := parquet.NewBuffer()
+		writer := parquet.NewWriter(output,
+			parquet.BloomFilters(
+				parquet.SplitBlockFilter("X"),
+				parquet.SplitBlockFilter("Y"),
+				parquet.SplitBlockFilter("Z"),
+			),
+		)
+		for i := range rows {
+			buffer.Write(&rows[i])
+		}
+		_, err := parquet.CopyRows(writer, buffer.Rows())
+		if err != nil {
+			t.Error(err)
+			return false
+		}
+		if err := writer.Close(); err != nil {
+			t.Error(err)
+			return false
+		}
+
+		reader := bytes.NewReader(output.Bytes())
+		f, err := parquet.OpenFile(reader, reader.Size())
+		if err != nil {
+			t.Error(err)
+			return false
+		}
+		rowGroup := f.RowGroup(0)
+		x := rowGroup.Column(0)
+		y := rowGroup.Column(1)
+		z := rowGroup.Column(2)
+
+		for i, col := range []parquet.ColumnChunk{x, y, z} {
+			if col.BloomFilter() == nil {
+				t.Errorf("column %d has no bloom filter despite being configured to have one", i)
+				return false
+			}
+		}
+
+		fx := x.BloomFilter()
+		fy := y.BloomFilter()
+		fz := z.BloomFilter()
+
+		test := func(f parquet.BloomFilter, v float64) bool {
+			b := [8]byte{}
+			binary.LittleEndian.PutUint64(b[:], math.Float64bits(v))
+			if ok, err := f.Check(b[:]); err != nil {
+				t.Errorf("unexpected error checking bloom filter: %v", err)
+				return false
+			} else if !ok {
+				t.Errorf("bloom filter does not contain value %g", v)
+				return false
+			}
+			return true
+		}
+
+		for _, row := range rows {
+			if !test(fx, row.X) || !test(fy, row.Y) || !test(fz, row.Z) {
+				return false
+			}
+		}
+
+		return true
+	}
+
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}
diff --git a/encoding/bytestreamsplit/decoder.go b/encoding/bytestreamsplit/decoder.go
index a595c1b..b0ecbf7 100644
--- a/encoding/bytestreamsplit/decoder.go
+++ b/encoding/bytestreamsplit/decoder.go
@@ -6,7 +6,6 @@ import (
 	"math"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type Decoder struct {
@@ -20,10 +19,6 @@ func NewDecoder(r io.Reader) *Decoder {
 	return &Decoder{reader: r}
 }
 
-func (d *Decoder) Encoding() format.Encoding {
-	return format.ByteStreamSplit
-}
-
 func (d *Decoder) Reset(r io.Reader) {
 	d.reader = r
 	d.offset = 0
diff --git a/encoding/bytestreamsplit/encoder.go b/encoding/bytestreamsplit/encoder.go
index a749485..9bc9953 100644
--- a/encoding/bytestreamsplit/encoder.go
+++ b/encoding/bytestreamsplit/encoder.go
@@ -5,7 +5,6 @@ import (
 	"math"
 
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/format"
 )
 
 type Encoder struct {
@@ -20,10 +19,6 @@ func NewEncoder(w io.Writer) *Encoder {
 	}
 }
 
-func (e *Encoder) Encoding() format.Encoding {
-	return format.ByteStreamSplit
-}
-
 func (e *Encoder) Write(b []byte) (int, error) {
 	return e.writer.Write(b)
 }
diff --git a/encoding/encoding_test.go b/encoding/encoding_test.go
index c58a0a3..2dfcd5e 100644
--- a/encoding/encoding_test.go
+++ b/encoding/encoding_test.go
@@ -248,11 +248,6 @@ func testEncoding(t *testing.T, e encoding.Encoding) {
 		scenario string
 		function func(*testing.T, encoding.Encoding)
 	}{
-		{
-			scenario: "encoding",
-			function: testFormatEncoding,
-		},
-
 		{
 			scenario: "boolean",
 			function: testBooleanEncoding,
@@ -307,19 +302,6 @@ func testEncoding(t *testing.T, e encoding.Encoding) {
 	}
 }
 
-func testFormatEncoding(t *testing.T, e encoding.Encoding) {
-	enc := e.NewEncoder(nil)
-	dec := e.NewDecoder(nil)
-
-	if enc.Encoding() != e.Encoding() {
-		t.Errorf("wrong encoder encoding: want=%s got=%s", e.Encoding(), enc.Encoding())
-	}
-
-	if dec.Encoding() != e.Encoding() {
-		t.Errorf("wrong decoder encoding: want=%s got=%s", e.Encoding(), dec.Encoding())
-	}
-}
-
 func testBooleanEncoding(t *testing.T, e encoding.Encoding) {
 	if !e.CanEncode(format.Boolean) {
 		t.Skipf("%s cannot encode boolean values", e)
diff --git a/writer_test.go b/writer_test.go
index 255f020..1af09de 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"strings"
 	"testing"
+	"testing/quick"
 
 	"github.com/hexops/gotextdiff"
 	"github.com/hexops/gotextdiff/myers"
@@ -459,3 +460,70 @@ func parquetTools(cmd, path string) ([]byte, error) {
 
 	return bytes.Join(lines, []byte("\n")), nil
 }
+
+func TestWriterGenerateBloomFilters(t *testing.T) {
+	type Person struct {
+		FirstName utf8string `parquet:"first_name"`
+		LastName  utf8string `parquet:"last_name"`
+	}
+
+	f := func(rows []Person) bool {
+		if len(rows) == 0 { // TODO: support writing files with no rows
+			return true
+		}
+
+		buffer := new(bytes.Buffer)
+		writer := parquet.NewWriter(buffer,
+			parquet.BloomFilters(
+				parquet.SplitBlockFilter("last_name"),
+			),
+		)
+		for i := range rows {
+			if err := writer.Write(&rows[i]); err != nil {
+				t.Error(err)
+				return false
+			}
+		}
+		if err := writer.Close(); err != nil {
+			t.Error(err)
+			return false
+		}
+
+		reader := bytes.NewReader(buffer.Bytes())
+		f, err := parquet.OpenFile(reader, reader.Size())
+		if err != nil {
+			t.Error(err)
+			return false
+		}
+		rowGroup := f.RowGroup(0)
+		firstName := rowGroup.Column(0)
+		lastName := rowGroup.Column(1)
+
+		if firstName.BloomFilter() != nil {
+			t.Errorf(`"first_name" column has a bloom filter even though none were configured`)
+			return false
+		}
+
+		bloomFilter := lastName.BloomFilter()
+		if bloomFilter == nil {
+			t.Error(`"last_name" column has no bloom filter despite being configured to have one`)
+			return false
+		}
+
+		for i, row := range rows {
+			if ok, err := bloomFilter.Check([]byte(row.LastName)); err != nil {
+				t.Errorf("unexpected error checking bloom filter: %v", err)
+				return false
+			} else if !ok {
+				t.Errorf("bloom filter does not contain value %q of row %d", row.LastName, i)
+				return false
+			}
+		}
+
+		return true
+	}
+
+	if err := quick.Check(f, nil); err != nil {
+		t.Error(err)
+	}
+}