Repository: segmentio/parquet-go
Pull request #432 (2022-12-06 20:51:10)

Remove parquet.Page.Clone
This PR removes the `Clone` method of the `Page` interface. This method was used as a mechanism to capture pages on columns that a writer needed to generate a bloom filter for, when we could not know ahead of time the size of the filter.

Cloning of the pages resulted in large memory utilization, which eventually contributed to reaching memory limits and causing processes to crash.

To work around this issue, we can instead trade off CPU to lower memory utilization, by generating the bloom filter from the list of pages encoded by the writer.


Patch:
diff --git a/convert.go b/convert.go
index dd23cba..200dd56 100644
--- a/convert.go
+++ b/convert.go
@@ -439,7 +439,6 @@ func (p missingPage) NumRows() int64                    { return p.numRows }
 func (p missingPage) NumValues() int64                  { return p.numValues }
 func (p missingPage) NumNulls() int64                   { return p.numNulls }
 func (p missingPage) Bounds() (min, max Value, ok bool) { return }
-func (p missingPage) Clone() Page                       { return p }
 func (p missingPage) Slice(i, j int64) Page             { return p }
 func (p missingPage) Size() int64                       { return 0 }
 func (p missingPage) RepetitionLevels() []byte          { return nil }
diff --git a/dictionary.go b/dictionary.go
index a3aa1a5..69c2f06 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -1297,14 +1297,6 @@ func (page *indexedPage) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *indexedPage) Clone() Page {
-	return &indexedPage{
-		typ:         page.typ,
-		values:      append([]int32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *indexedPage) Slice(i, j int64) Page {
 	return &indexedPage{
 		typ:         page.typ,
diff --git a/page.go b/page.go
index 883e7c5..12f0b0a 100644
--- a/page.go
+++ b/page.go
@@ -62,10 +62,6 @@ type Page interface {
 	// available.
 	Values() ValueReader
 
-	// Returns a copy of the page which does not share any of the buffers, but
-	// contains the same values, repetition and definition levels.
-	Clone() Page
-
 	// Returns a new page which is as slice of the receiver between row indexes
 	// i and j.
 	Slice(i, j int64) Page
@@ -93,13 +89,6 @@ type Page interface {
 type PageReader interface {
 	// Reads and returns the next page from the sequence. When all pages have
 	// been read, or if the sequence was closed, the method returns io.EOF.
-	//
-	// The returned page and other objects derived from it remain valid until
-	// the next call to ReadPage, or until the sequence is closed. The page
-	// reader may use this property to optimize resource management by reusing
-	// memory across pages. Applications that need to acquire ownership of the
-	// returned page must clone by calling page.Buffer().Clone() to create a
-	// copy in memory.
 	ReadPage() (Page, error)
 }
 
@@ -323,7 +312,6 @@ func (page *errorPage) NumRows() int64                    { return 1 }
 func (page *errorPage) NumValues() int64                  { return 1 }
 func (page *errorPage) NumNulls() int64                   { return 0 }
 func (page *errorPage) Bounds() (min, max Value, ok bool) { return }
-func (page *errorPage) Clone() Page                       { return page }
 func (page *errorPage) Slice(i, j int64) Page             { return page }
 func (page *errorPage) Size() int64                       { return 1 }
 func (page *errorPage) RepetitionLevels() []byte          { return nil }
@@ -385,14 +373,6 @@ func (page *optionalPage) Values() ValueReader {
 	}
 }
 
-func (page *optionalPage) Clone() Page {
-	return newOptionalPage(
-		page.base.Clone(),
-		page.maxDefinitionLevel,
-		copyBytes(page.definitionLevels),
-	)
-}
-
 func (page *optionalPage) Slice(i, j int64) Page {
 	maxDefinitionLevel := page.maxDefinitionLevel
 	definitionLevels := page.definitionLevels
@@ -456,16 +436,6 @@ func (page *repeatedPage) Values() ValueReader {
 	}
 }
 
-func (page *repeatedPage) Clone() Page {
-	return newRepeatedPage(
-		page.base.Clone(),
-		page.maxRepetitionLevel,
-		page.maxDefinitionLevel,
-		copyBytes(page.repetitionLevels),
-		copyBytes(page.definitionLevels),
-	)
-}
-
 func (page *repeatedPage) Slice(i, j int64) Page {
 	numRows := page.NumRows()
 	if i < 0 || i > numRows {
@@ -614,16 +584,6 @@ func (page *booleanPage) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *booleanPage) Clone() Page {
-	return &booleanPage{
-		typ:         page.typ,
-		bits:        append([]byte{}, page.bits...),
-		offset:      page.offset,
-		numValues:   page.numValues,
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *booleanPage) Slice(i, j int64) Page {
 	off := i / 8
 	end := j / 8
@@ -698,14 +658,6 @@ func (page *int32Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *int32Page) Clone() Page {
-	return &int32Page{
-		typ:         page.typ,
-		values:      append([]int32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *int32Page) Slice(i, j int64) Page {
 	return &int32Page{
 		typ:         page.typ,
@@ -771,14 +723,6 @@ func (page *int64Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *int64Page) Clone() Page {
-	return &int64Page{
-		typ:         page.typ,
-		values:      append([]int64{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *int64Page) Slice(i, j int64) Page {
 	return &int64Page{
 		typ:         page.typ,
@@ -846,14 +790,6 @@ func (page *int96Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *int96Page) Clone() Page {
-	return &int96Page{
-		typ:         page.typ,
-		values:      append([]deprecated.Int96{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *int96Page) Slice(i, j int64) Page {
 	return &int96Page{
 		typ:         page.typ,
@@ -919,14 +855,6 @@ func (page *floatPage) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *floatPage) Clone() Page {
-	return &floatPage{
-		typ:         page.typ,
-		values:      append([]float32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *floatPage) Slice(i, j int64) Page {
 	return &floatPage{
 		typ:         page.typ,
@@ -992,14 +920,6 @@ func (page *doublePage) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *doublePage) Clone() Page {
-	return &doublePage{
-		typ:         page.typ,
-		values:      append([]float64{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *doublePage) Slice(i, j int64) Page {
 	return &doublePage{
 		typ:         page.typ,
@@ -1133,15 +1053,6 @@ func (page *byteArrayPage) cloneOffsets() []uint32 {
 	return offsets
 }
 
-func (page *byteArrayPage) Clone() Page {
-	return &byteArrayPage{
-		typ:         page.typ,
-		values:      page.cloneValues(),
-		offsets:     page.cloneOffsets(),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *byteArrayPage) Slice(i, j int64) Page {
 	return &byteArrayPage{
 		typ:         page.typ,
@@ -1223,15 +1134,6 @@ func (page *fixedLenByteArrayPage) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *fixedLenByteArrayPage) Clone() Page {
-	return &fixedLenByteArrayPage{
-		typ:         page.typ,
-		data:        append([]byte{}, page.data...),
-		size:        page.size,
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *fixedLenByteArrayPage) Slice(i, j int64) Page {
 	return &fixedLenByteArrayPage{
 		typ:         page.typ,
@@ -1304,14 +1206,6 @@ func (page *uint32Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *uint32Page) Clone() Page {
-	return &uint32Page{
-		typ:         page.typ,
-		values:      append([]uint32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *uint32Page) Slice(i, j int64) Page {
 	return &uint32Page{
 		typ:         page.typ,
@@ -1377,14 +1271,6 @@ func (page *uint64Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *uint64Page) Clone() Page {
-	return &uint64Page{
-		typ:         page.typ,
-		values:      append([]uint64{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *uint64Page) Slice(i, j int64) Page {
 	return &uint64Page{
 		typ:         page.typ,
@@ -1450,14 +1336,6 @@ func (page *be128Page) Bounds() (min, max Value, ok bool) {
 	return min, max, ok
 }
 
-func (page *be128Page) Clone() Page {
-	return &be128Page{
-		typ:         page.typ,
-		values:      append([][16]byte{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
 func (page *be128Page) Slice(i, j int64) Page {
 	return &be128Page{
 		typ:         page.typ,
@@ -1507,7 +1385,6 @@ func (page *nullPage) Size() int64                       { return 1 }
 func (page *nullPage) Values() ValueReader {
 	return &nullPageValues{column: page.column, remain: page.count}
 }
-func (page *nullPage) Clone() Page { return page }
 func (page *nullPage) Slice(i, j int64) Page {
 	return &nullPage{column: page.column, count: page.count - int(j-i)}
 }
diff --git a/page_buffer.go b/page_buffer.go
index 8dcd158..ba67245 100644
--- a/page_buffer.go
+++ b/page_buffer.go
@@ -1,7 +1,7 @@
 package parquet
 
 import (
-	"bytes"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -22,9 +22,9 @@ import (
 // PageBufferPool implementations must be safe to use concurrently from multiple
 // goroutines.
 type PageBufferPool interface {
-	// GetPageBuffer is called when a parquet writer needs to acquires a new
+	// GetPageBuffer is called when a parquet writer needs to acquire a new
 	// page buffer from the pool.
-	GetPageBuffer() io.ReadWriter
+	GetPageBuffer() io.ReadWriteSeeker
 
 	// PutPageBuffer is called when a parquet writer releases a page buffer to
 	// the pool.
@@ -32,7 +32,7 @@ type PageBufferPool interface {
 	// The parquet.Writer type guarantees that the buffers it calls this method
 	// with were previously acquired by a call to GetPageBuffer on the same
 	// pool, and that it will not use them anymore after the call.
-	PutPageBuffer(io.ReadWriter)
+	PutPageBuffer(io.ReadWriteSeeker)
 }
 
 // NewPageBufferPool creates a new in-memory page buffer pool.
@@ -41,20 +41,73 @@ type PageBufferPool interface {
 // Go heap.
 func NewPageBufferPool() PageBufferPool { return new(pageBufferPool) }
 
+type pageBuffer struct {
+	data []byte
+	off  int
+}
+
+func (p *pageBuffer) Reset() {
+	p.data, p.off = p.data[:0], 0
+}
+
+func (p *pageBuffer) Read(b []byte) (n int, err error) {
+	n = copy(b, p.data[p.off:])
+	p.off += n
+	if p.off == len(p.data) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (p *pageBuffer) Write(b []byte) (int, error) {
+	n := copy(p.data[p.off:cap(p.data)], b)
+	p.data = p.data[:p.off+n]
+
+	if n < len(b) {
+		p.data = append(p.data, b[n:]...)
+	}
+
+	p.off += len(b)
+	return len(b), nil
+}
+
+func (p *pageBuffer) WriteTo(w io.Writer) (int64, error) {
+	n, err := w.Write(p.data[p.off:])
+	p.off += n
+	return int64(n), err
+}
+
+func (p *pageBuffer) Seek(offset int64, whence int) (int64, error) {
+	switch whence {
+	case io.SeekCurrent:
+		offset += int64(p.off)
+	case io.SeekEnd:
+		offset += int64(len(p.data))
+	}
+	if offset < 0 {
+		return 0, fmt.Errorf("seek: negative offset: %d<0", offset)
+	}
+	if offset > int64(len(p.data)) {
+		offset = int64(len(p.data))
+	}
+	p.off = int(offset)
+	return offset, nil
+}
+
 type pageBufferPool struct{ sync.Pool }
 
-func (pool *pageBufferPool) GetPageBuffer() io.ReadWriter {
-	b, _ := pool.Get().(*bytes.Buffer)
+func (pool *pageBufferPool) GetPageBuffer() io.ReadWriteSeeker {
+	b, _ := pool.Get().(*pageBuffer)
 	if b == nil {
-		b = new(bytes.Buffer)
+		b = new(pageBuffer)
 	} else {
 		b.Reset()
 	}
 	return b
 }
 
-func (pool *pageBufferPool) PutPageBuffer(buf io.ReadWriter) {
-	if b, _ := buf.(*bytes.Buffer); b != nil {
+func (pool *pageBufferPool) PutPageBuffer(buf io.ReadWriteSeeker) {
+	if b, _ := buf.(*pageBuffer); b != nil {
 		pool.Put(b)
 	}
 }
@@ -75,7 +128,7 @@ func NewFileBufferPool(tempdir, pattern string) PageBufferPool {
 	return pool
 }
 
-func (pool *fileBufferPool) GetPageBuffer() io.ReadWriter {
+func (pool *fileBufferPool) GetPageBuffer() io.ReadWriteSeeker {
 	if pool.err != nil {
 		return &errorBuffer{err: pool.err}
 	}
@@ -83,58 +136,28 @@ func (pool *fileBufferPool) GetPageBuffer() io.ReadWriter {
 	if err != nil {
 		return &errorBuffer{err: err}
 	}
-	return &fileBuffer{file: f}
+	return f
 }
 
-func (pool *fileBufferPool) PutPageBuffer(buf io.ReadWriter) {
-	if f, _ := buf.(*fileBuffer); f != nil {
-		defer f.file.Close()
-		os.Remove(f.file.Name())
+func (pool *fileBufferPool) PutPageBuffer(buf io.ReadWriteSeeker) {
+	if f, _ := buf.(*os.File); f != nil {
+		defer f.Close()
+		os.Remove(f.Name())
 	}
 }
 
-type fileBuffer struct {
-	file *os.File
-	seek int64
-}
-
-func (buf *fileBuffer) Read(b []byte) (int, error) {
-	// The *os.File tracks a single cursor which we use for write operations to
-	// support appending to the buffer. We need a second cursor for reads which
-	// is tracked by the buf.seek field, using ReadAt to read from the file at
-	// the current read position.
-	n, err := buf.file.ReadAt(b, buf.seek)
-	buf.seek += int64(n)
-	return n, err
-}
-
-func (buf *fileBuffer) ReadFrom(r io.Reader) (int64, error) {
-	return buf.file.ReadFrom(r)
-}
-
-func (buf *fileBuffer) Write(b []byte) (int, error) {
-	return buf.file.Write(b)
-}
-
-func (buf *fileBuffer) WriteString(s string) (int, error) {
-	return buf.file.WriteString(s)
-}
-
 type errorBuffer struct{ err error }
 
 func (buf *errorBuffer) Read([]byte) (int, error)          { return 0, buf.err }
 func (buf *errorBuffer) Write([]byte) (int, error)         { return 0, buf.err }
-func (buf *errorBuffer) WriteString(string) (int, error)   { return 0, buf.err }
 func (buf *errorBuffer) ReadFrom(io.Reader) (int64, error) { return 0, buf.err }
 func (buf *errorBuffer) WriteTo(io.Writer) (int64, error)  { return 0, buf.err }
+func (buf *errorBuffer) Seek(int64, int) (int64, error)    { return 0, buf.err }
 
 var (
 	defaultPageBufferPool pageBufferPool
 
-	_ io.ReaderFrom   = (*fileBuffer)(nil)
-	_ io.StringWriter = (*fileBuffer)(nil)
-
-	_ io.ReaderFrom   = (*errorBuffer)(nil)
-	_ io.WriterTo     = (*errorBuffer)(nil)
-	_ io.StringWriter = (*errorBuffer)(nil)
+	_ io.ReaderFrom = (*errorBuffer)(nil)
+	_ io.WriterTo   = (*errorBuffer)(nil)
+	_ io.WriterTo   = (*pageBuffer)(nil)
 )
diff --git a/writer.go b/writer.go
index 6bb76fd..6c4c911 100644
--- a/writer.go
+++ b/writer.go
@@ -382,8 +382,8 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 			c.encodings = addEncoding(c.encodings, format.Plain)
 		}
 
-		c.page.encoding = encoding
-		c.encodings = addEncoding(c.encodings, c.page.encoding.Encoding())
+		c.encoding = encoding
+		c.encodings = addEncoding(c.encodings, c.encoding.Encoding())
 		sortPageEncodings(c.encodings)
 
 		w.columns = append(w.columns, c)
@@ -602,7 +602,7 @@ func (w *writer) writeRowGroup(rowGroupSchema *Schema, rowGroupSortingColumns []
 	fileOffset := w.writer.offset
 
 	for _, c := range w.columns {
-		if len(c.filter.bits) > 0 {
+		if len(c.filter) > 0 {
 			c.columnChunk.MetaData.BloomFilterOffset = w.writer.offset
 			if err := c.writeBloomFilter(&w.writer); err != nil {
 				return 0, err
@@ -846,13 +846,14 @@ func (wb *writerBuffers) swapPageAndScratchBuffers() {
 
 type writerColumn struct {
 	pool  PageBufferPool
-	pages []io.ReadWriter
+	pages []io.ReadWriteSeeker
 
 	columnPath   columnPath
 	columnType   Type
 	columnIndex  ColumnIndexer
 	columnBuffer ColumnBuffer
 	columnFilter BloomFilterColumn
+	encoding     encoding.Encoding
 	compression  compress.Codec
 	dictionary   Dictionary
 
@@ -867,15 +868,7 @@ type writerColumn struct {
 		encoder  thrift.Encoder
 	}
 
-	page struct {
-		encoding encoding.Encoding
-	}
-
-	filter struct {
-		bits  []byte
-		pages []Page
-	}
-
+	filter         []byte
 	numRows        int64
 	maxValues      int32
 	numValues      int32
@@ -905,14 +898,10 @@ func (c *writerColumn) reset() {
 	for i := range c.pages {
 		c.pages[i] = nil
 	}
-	for i := range c.filter.pages {
-		c.filter.pages[i] = nil
-	}
 	c.pages = c.pages[:0]
 	// Bloom filters may change in size between row groups, but we retain the
 	// buffer to avoid reallocating large memory blocks.
-	c.filter.bits = c.filter.bits[:0]
-	c.filter.pages = c.filter.pages[:0]
+	c.filter = c.filter[:0]
 	c.numRows = 0
 	c.numValues = 0
 	// Reset the fields of column chunks that change between row groups,
@@ -949,44 +938,109 @@ func (c *writerColumn) flush() (err error) {
 }
 
 func (c *writerColumn) flushFilterPages() error {
-	if c.columnFilter != nil {
-		// If there is a dictionary, it contains all the values that we need to
-		// write to the filter.
-		if dict := c.dictionary; dict != nil {
-			// Need to always attempt to resize the filter, as the writer might
-			// be reused after resetting which would have reset the length of
-			// the filter to 0.
-			c.resizeBloomFilter(int64(dict.Len()))
-			if err := c.writePageToFilter(dict.Page()); err != nil {
-				return err
-			}
+	if c.columnFilter == nil {
+		return nil
+	}
+
+	// If there is a dictionary, it contains all the values that we need to
+	// write to the filter.
+	if dict := c.dictionary; dict != nil {
+		// Need to always attempt to resize the filter, as the writer might
+		// be reused after resetting which would have reset the length of
+		// the filter to 0.
+		c.resizeBloomFilter(int64(dict.Len()))
+		return c.writePageToFilter(dict.Page())
+	}
+
+	// When the filter was already allocated, pages have been written to it as
+	// they were seen by the column writer.
+	if len(c.filter) > 0 {
+		return nil
+	}
+
+	// When the filter was not allocated, the writer did not know how many
+	// values were going to be seen and therefore could not properly size the
+	// filter ahead of time. In this case, we read back all the pages that we
+	// have encoded and copy their values back to the filter.
+	//
+	// A prior implementation of the column writer used to create in-memory
+	// copies of the pages to avoid this decoding step; however, this unbounded
+	// allocation caused memory exhaustion in production applications. CPU being
+	// a somewhat more stretchable resource, we prefer spending time on this
+	// decoding step than having to trigger incident response when production
+	// systems are getting OOM-Killed.
+	c.resizeBloomFilter(c.columnChunk.MetaData.NumValues)
+
+	column := &Column{
+		// Set all the fields required by the decodeDataPage* methods.
+		typ:                c.columnType,
+		encoding:           c.encoding,
+		compression:        c.compression,
+		maxRepetitionLevel: c.maxRepetitionLevel,
+		maxDefinitionLevel: c.maxDefinitionLevel,
+		index:              int16(c.bufferIndex),
+	}
+
+	rbuf, pool := getBufioReader(nil, 1024)
+	pbuf := (*buffer)(nil)
+	defer func() {
+		putBufioReader(rbuf, pool)
+		if pbuf != nil {
+			buffers.put(pbuf)
 		}
+	}()
 
-		if len(c.filter.pages) > 0 {
-			numValues := int64(0)
-			for _, page := range c.filter.pages {
-				numValues += page.NumValues()
-			}
-			c.resizeBloomFilter(numValues)
-			for _, page := range c.filter.pages {
-				if err := c.writePageToFilter(page); err != nil {
-					return err
-				}
-			}
+	decoder := thrift.NewDecoder(c.header.protocol.NewReader(rbuf))
+
+	for _, p := range c.pages {
+		rbuf.Reset(p)
+
+		header := new(format.PageHeader)
+		if err := decoder.Decode(header); err != nil {
+			return err
+		}
+
+		if pbuf != nil {
+			buffers.put(pbuf)
+		}
+		pbuf = buffers.get(int(header.CompressedPageSize))
+		if _, err := io.ReadFull(rbuf, pbuf.data); err != nil {
+			return err
+		}
+		if _, err := p.Seek(0, io.SeekStart); err != nil {
+			return err
+		}
+
+		var page Page
+		var err error
+
+		switch header.Type {
+		case format.DataPage:
+			page, err = column.decodeDataPageV1(DataPageHeaderV1{header.DataPageHeader}, pbuf, nil, header.UncompressedPageSize)
+		case format.DataPageV2:
+			page, err = column.decodeDataPageV2(DataPageHeaderV2{header.DataPageHeaderV2}, pbuf, nil, header.UncompressedPageSize)
+		}
+		if page != nil {
+			err = c.writePageToFilter(page)
+			Release(page)
+		}
+		if err != nil {
+			return err
 		}
 	}
+
 	return nil
 }
 
 func (c *writerColumn) resizeBloomFilter(numValues int64) {
 	const bitsPerValue = 10 // TODO: make this configurable
 	filterSize := c.columnFilter.Size(numValues, bitsPerValue)
-	if cap(c.filter.bits) < filterSize {
-		c.filter.bits = make([]byte, filterSize)
+	if cap(c.filter) < filterSize {
+		c.filter = make([]byte, filterSize)
 	} else {
-		c.filter.bits = c.filter.bits[:filterSize]
-		for i := range c.filter.bits {
-			c.filter.bits[i] = 0
+		c.filter = c.filter[:filterSize]
+		for i := range c.filter {
+			c.filter[i] = 0
 		}
 	}
 }
@@ -1039,11 +1093,11 @@ func (c *writerColumn) WriteValues(values []Value) (numValues int, err error) {
 func (c *writerColumn) writeBloomFilter(w io.Writer) error {
 	e := thrift.NewEncoder(c.header.protocol.NewWriter(w))
 	h := bloomFilterHeader(c.columnFilter)
-	h.NumBytes = int32(len(c.filter.bits))
+	h.NumBytes = int32(len(c.filter))
 	if err := e.Encode(&h); err != nil {
 		return err
 	}
-	_, err := w.Write(c.filter.bits)
+	_, err := w.Write(c.filter)
 	return err
 }
 
@@ -1063,7 +1117,7 @@ func (c *writerColumn) writeDataPage(page Page) (int64, error) {
 		buf.encodeDefinitionLevels(page, c.maxDefinitionLevel)
 	}
 
-	if err := buf.encode(page, c.page.encoding); err != nil {
+	if err := buf.encode(page, c.encoding); err != nil {
 		return 0, fmt.Errorf("encoding parquet data page: %w", err)
 	}
 	if c.dataPageType == format.DataPage {
@@ -1077,24 +1131,14 @@ func (c *writerColumn) writeDataPage(page Page) (int64, error) {
 		}
 	}
 
-	if page.Dictionary() == nil {
-		switch {
-		case len(c.filter.bits) > 0:
-			// When the writer knows the number of values in advance (e.g. when
-			// writing a full row group), the filter encoding is set and the page
-			// can be directly applied to the filter, which minimizes memory usage
-			// since there is no need to buffer the values in order to determine
-			// the size of the filter.
-			if err := c.writePageToFilter(page); err != nil {
-				return 0, err
-			}
-		case c.columnFilter != nil:
-			// If the column uses a dictionary encoding, all possible values exist
-			// in the dictionary and there is no need to buffer the pages, but if
-			// the column is supposed to generate a filter and the number of values
-			// wasn't known, we must buffer all the pages in order to properly size
-			// the filter.
-			c.filter.pages = append(c.filter.pages, page.Clone())
+	if page.Dictionary() == nil && len(c.filter) > 0 {
+		// When the writer knows the number of values in advance (e.g. when
+		// writing a full row group), the filter encoding is set and the page
+		// can be directly applied to the filter, which minimizes memory usage
+		// since there is no need to buffer the values in order to determine
+		// the size of the filter.
+		if err := c.writePageToFilter(page); err != nil {
+			return 0, err
 		}
 	}
 
@@ -1116,7 +1160,7 @@ func (c *writerColumn) writeDataPage(page Page) (int64, error) {
 	case format.DataPage:
 		pageHeader.DataPageHeader = &format.DataPageHeader{
 			NumValues:               int32(numValues),
-			Encoding:                c.page.encoding.Encoding(),
+			Encoding:                c.encoding.Encoding(),
 			DefinitionLevelEncoding: format.RLE,
 			RepetitionLevelEncoding: format.RLE,
 			Statistics:              statistics,
@@ -1126,7 +1170,7 @@ func (c *writerColumn) writeDataPage(page Page) (int64, error) {
 			NumValues:                  int32(numValues),
 			NumNulls:                   int32(numNulls),
 			NumRows:                    int32(numRows),
-			Encoding:                   c.page.encoding.Encoding(),
+			Encoding:                   c.encoding.Encoding(),
 			DefinitionLevelsByteLength: int32(len(buf.definitions)),
 			RepetitionLevelsByteLength: int32(len(buf.repetitions)),
 			IsCompressed:               &c.isCompressed,
@@ -1212,7 +1256,7 @@ func (c *writerColumn) writeDictionaryPage(output io.Writer, dict Dictionary) (e
 func (w *writerColumn) writePageToFilter(page Page) (err error) {
 	pageType := page.Type()
 	pageData := page.Data()
-	w.filter.bits, err = pageType.Encode(w.filter.bits, pageData, w.columnFilter.Encoding())
+	w.filter, err = pageType.Encode(w.filter, pageData, w.columnFilter.Encoding())
 	return err
 }
 
@@ -1230,6 +1274,13 @@ func (c *writerColumn) writePageTo(size int64, writeTo func(io.Writer) (int64, e
 	if written != size {
 		return fmt.Errorf("writing parquet column page expected %dB but got %dB: %w", size, written, io.ErrShortWrite)
 	}
+	offset, err := buffer.Seek(0, io.SeekStart)
+	if err != nil {
+		return err
+	}
+	if offset != 0 {
+		return fmt.Errorf("resetting parquet page buffer to the start expected offset zero but got %d", offset)
+	}
 	c.pages, buffer = append(c.pages, buffer), nil
 	return nil
 }


Test patch:
diff --git a/page_buffer_test.go b/page_buffer_test.go
index 6dd145b..6166640 100644
--- a/page_buffer_test.go
+++ b/page_buffer_test.go
@@ -93,6 +93,9 @@ func testPageBufferPoolCopyFromBuffer(t *testing.T, pool parquet.PageBufferPool)
 	if _, err := io.WriteString(buffer, content); err != nil {
 		t.Fatal(err)
 	}
+	if _, err := buffer.Seek(0, io.SeekStart); err != nil {
+		t.Fatal(err)
+	}
 
 	writer := new(bytes.Buffer)
 	_, err := io.Copy(struct{ io.Writer }{writer}, buffer)
@@ -100,13 +103,20 @@ func testPageBufferPoolCopyFromBuffer(t *testing.T, pool parquet.PageBufferPool)
 		t.Fatal(err)
 	}
 
-	assertBufferContent(t, writer, content)
+	assertBufferContent(t, bytes.NewReader(writer.Bytes()), content)
 }
 
-func assertBufferContent(t *testing.T, b io.Reader, s string) {
+func assertBufferContent(t *testing.T, b io.ReadSeeker, s string) {
 	t.Helper()
 
+	offset, err := b.Seek(0, io.SeekStart)
+	if err != nil {
+		t.Error("seek:", err)
+	}
+	if offset != 0 {
+		t.Errorf("seek: invalid offset returned: want=0 got=%d", offset)
+	}
 	if err := iotest.TestReader(b, []byte(s)); err != nil {
-		t.Error(err)
+		t.Error("iotest:", err)
 	}
 }