Repository: segmentio/parquet-go
Pull request #163 (2022-04-29 19:46:32)

Refactor compression APIs
As I was working on https://github.com/segmentio/parquet-go/pull/161, it became apparent that a lot of the complexity of reading files came from the way the compression APIs were based on reading and writing byte streams, despite pages always being manipulated in memory.

With this PR, I am attempting to reduce this complexity to enable further optimizations by changing the compression models to stateless `Encode`/`Decode` methods instead of the `NewReader`/`NewWriter` we had previously.

While this is a breaking change, I don't expect this to impact users much since the compression codecs are managed internally, programs do not need to use them directly.


Patch:
diff --git a/column_buffer.go b/column_buffer.go
index 0631f9b..0a6fb07 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -447,8 +447,9 @@ func (col *repeatedColumnBuffer) Page() BufferedPage {
 
 		column := col.reordering
 		column.Reset()
+		maxNumValues := 0
 		defer func() {
-			clearValues(col.buffer[:cap(col.buffer)])
+			clearValues(col.buffer[:maxNumValues])
 		}()
 
 		baseOffset := 0
@@ -472,6 +473,9 @@ func (col *repeatedColumnBuffer) Page() BufferedPage {
 				if _, err := column.base.WriteValues(col.buffer); err != nil {
 					return newErrorPage(col.Column(), "reordering rows of repeated column: %w", err)
 				}
+				if numValues > maxNumValues {
+					maxNumValues = numValues
+				}
 			}
 
 			column.rows = append(column.rows, region{
@@ -572,8 +576,9 @@ func (col *repeatedColumnBuffer) WriteValues(values []Value) (numValues int, err
 		}
 	}
 
+	maxNumValues := 0
 	defer func() {
-		clearValues(col.buffer[:cap(col.buffer)])
+		clearValues(col.buffer[:maxNumValues])
 	}()
 
 	var row []Value
@@ -584,6 +589,9 @@ func (col *repeatedColumnBuffer) WriteValues(values []Value) (numValues int, err
 			return numValues, err
 		}
 		numValues += len(row)
+		if len(col.buffer) > maxNumValues {
+			maxNumValues = len(col.buffer)
+		}
 	}
 
 	return numValues, nil
diff --git a/compress.go b/compress.go
index 88b8dcb..2757878 100644
--- a/compress.go
+++ b/compress.go
@@ -2,10 +2,6 @@ package parquet
 
 import (
 	"fmt"
-	"io"
-	"runtime"
-	"sort"
-	"sync"
 
 	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/compress/brotli"
@@ -38,8 +34,7 @@ var (
 
 	// Zstd is the ZSTD parquet compression codec.
 	Zstd = zstd.Codec{
-		Level:       zstd.DefaultLevel,
-		Concurrency: zstd.DefaultConcurrency,
+		Level: zstd.DefaultLevel,
 	}
 
 	// Lz4Raw is the LZ4_RAW parquet compression codec.
@@ -56,11 +51,6 @@ var (
 		format.Zstd:         &Zstd,
 		format.Lz4Raw:       &Lz4Raw,
 	}
-
-	// Pools of compressed page readers used to retain compression codecs across
-	// page reads to reduce the compute and memory footprint of creating new
-	// decompressors for every new page read in a parquet file.
-	compressedPageReaders = [len(compressionCodecs)]sync.Pool{}
 )
 
 // LookupCompressionCodec returns the compression codec associated with the
@@ -77,46 +67,10 @@ func LookupCompressionCodec(codec format.CompressionCodec) compress.Codec {
 	return &unsupported{codec}
 }
 
-func acquireCompressedPageReader(codec format.CompressionCodec, page io.Reader) *compressedPageReader {
-	r, _ := compressedPageReaders[codec].Get().(*compressedPageReader)
-	if r == nil {
-		r = &compressedPageReader{codec: codec}
-		r.reader, r.err = LookupCompressionCodec(codec).NewReader(page)
-		runtime.SetFinalizer(r, func(r *compressedPageReader) { r.Close() })
-	} else {
-		r.Reset(page)
-	}
-	return r
-}
-
-func releaseCompressedPageReader(r *compressedPageReader) {
-	r.Reset(nil)
-	compressedPageReaders[r.codec].Put(r)
-}
-
-type compressedPageReader struct {
-	codec  format.CompressionCodec
-	reader compress.Reader
-	err    error
-}
-
-func (r *compressedPageReader) Close() error {
-	return r.reader.Close()
+type unsupported struct {
+	codec format.CompressionCodec
 }
 
-func (r *compressedPageReader) Read(b []byte) (int, error) {
-	if r.err != nil {
-		return 0, r.err
-	}
-	return r.reader.Read(b)
-}
-
-func (r *compressedPageReader) Reset(page io.Reader) {
-	r.err = r.reader.Reset(page)
-}
-
-type unsupported struct{ codec format.CompressionCodec }
-
 func (u *unsupported) String() string {
 	return "UNSUPPORTED"
 }
@@ -125,56 +79,14 @@ func (u *unsupported) CompressionCodec() format.CompressionCodec {
 	return u.codec
 }
 
-func (u *unsupported) NewReader(r io.Reader) (compress.Reader, error) {
-	return unsupportedReader{u}, nil
+func (u *unsupported) Encode(dst, src []byte) ([]byte, error) {
+	return dst[:0], u.error()
 }
 
-func (u *unsupported) NewWriter(w io.Writer) (compress.Writer, error) {
-	return unsupportedWriter{u}, nil
+func (u *unsupported) Decode(dst, src []byte) ([]byte, error) {
+	return dst[:0], u.error()
 }
 
 func (u *unsupported) error() error {
 	return fmt.Errorf("unsupported compression codec: %s", u.codec)
 }
-
-type unsupportedReader struct{ *unsupported }
-
-func (r unsupportedReader) Close() error               { return nil }
-func (r unsupportedReader) Reset(io.Reader) error      { return nil }
-func (r unsupportedReader) Read(b []byte) (int, error) { return 0, r.error() }
-
-type unsupportedWriter struct{ *unsupported }
-
-func (w unsupportedWriter) Close() error                { return nil }
-func (w unsupportedWriter) Flush() error                { return nil }
-func (w unsupportedWriter) Reset(io.Writer) error       { return nil }
-func (w unsupportedWriter) Write(b []byte) (int, error) { return 0, w.error() }
-
-func sortCodecs(codecs []compress.Codec) {
-	if len(codecs) > 1 {
-		sort.Slice(codecs, func(i, j int) bool {
-			return codecs[i].CompressionCodec() < codecs[j].CompressionCodec()
-		})
-	}
-}
-
-func dedupeSortedCodecs(codecs []compress.Codec) []compress.Codec {
-	if len(codecs) > 1 {
-		i := 0
-
-		for _, c := range codecs[1:] {
-			if c.CompressionCodec() != codecs[i].CompressionCodec() {
-				i++
-				codecs[i] = c
-			}
-		}
-
-		clear := codecs[i+1:]
-		for i := range clear {
-			clear[i] = nil
-		}
-
-		codecs = codecs[:i+1]
-	}
-	return codecs
-}
diff --git a/compress/brotli/brotli.go b/compress/brotli/brotli.go
index 8679eb2..c2f6c2f 100644
--- a/compress/brotli/brotli.go
+++ b/compress/brotli/brotli.go
@@ -21,6 +21,9 @@ type Codec struct {
 	// LGWin is the base 2 logarithm of the sliding window size.
 	// Range is 10 to 24. 0 indicates automatic configuration based on Quality.
 	LGWin int
+
+	r compress.Decompressor
+	w compress.Compressor
 }
 
 func (c *Codec) String() string {
@@ -31,22 +34,21 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Brotli
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	return reader{brotli.NewReader(r)}, nil
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	return c.w.Encode(dst, src, func(w io.Writer) (compress.Writer, error) {
+		return brotli.NewWriterOptions(w, brotli.WriterOptions{
+			Quality: c.Quality,
+			LGWin:   c.LGWin,
+		}), nil
+	})
 }
 
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	opts := brotli.WriterOptions{
-		Quality: c.Quality,
-		LGWin:   c.LGWin,
-	}
-	return writer{brotli.NewWriterOptions(w, opts)}, nil
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	return c.r.Decode(dst, src, func(r io.Reader) (compress.Reader, error) {
+		return reader{brotli.NewReader(r)}, nil
+	})
 }
 
 type reader struct{ *brotli.Reader }
 
-func (r reader) Close() error { return nil }
-
-type writer struct{ *brotli.Writer }
-
-func (w writer) Reset(ww io.Writer) error { w.Writer.Reset(ww); return nil }
+func (reader) Close() error { return nil }
diff --git a/compress/compress.go b/compress/compress.go
index b6cfa98..c1ae918 100644
--- a/compress/compress.go
+++ b/compress/compress.go
@@ -5,7 +5,9 @@
 package compress
 
 import (
+	"bytes"
 	"io"
+	"sync"
 
 	"github.com/segmentio/parquet-go/format"
 )
@@ -21,25 +23,120 @@ type Codec interface {
 	// Returns the code of the compression codec in the parquet format.
 	CompressionCodec() format.CompressionCodec
 
-	// Creates a new Reader decompressing data from the io.Reader passed as
-	// argument.
-	NewReader(io.Reader) (Reader, error)
+	// Writes the compressed version of src to dst and returns it.
+	//
+	// The method automatically reallocates the output buffer if its capacity
+	// was too small to hold the compressed data.
+	Encode(dst, src []byte) ([]byte, error)
 
-	// Creates a new Writer compressing data to the io.Writer passed as
-	// argument.
-	NewWriter(io.Writer) (Writer, error)
+	// Writes the uncompressed version of src to dst and returns it.
+	//
+	// The method automatically reallocates the output buffer if its capacity
+	// was too small to hold the uncompressed data.
+	Decode(dst, src []byte) ([]byte, error)
 }
 
-// Reader is an extension of io.Reader implemented by all decompressors.
 type Reader interface {
-	io.Reader
-	io.Closer
+	io.ReadCloser
 	Reset(io.Reader) error
 }
 
-// Writer is an extension of io.Writer implemented by all compressors.
 type Writer interface {
-	io.Writer
-	io.Closer
-	Reset(io.Writer) error
+	io.WriteCloser
+	Reset(io.Writer)
+}
+
+type Compressor struct {
+	writers sync.Pool // *writer
+}
+
+type writer struct {
+	output bytes.Buffer
+	writer Writer
+}
+
+func (c *Compressor) Encode(dst, src []byte, newWriter func(io.Writer) (Writer, error)) ([]byte, error) {
+	w, _ := c.writers.Get().(*writer)
+	if w != nil {
+		w.output = *bytes.NewBuffer(dst[:0])
+		w.writer.Reset(&w.output)
+	} else {
+		w = new(writer)
+		w.output = *bytes.NewBuffer(dst[:0])
+		var err error
+		if w.writer, err = newWriter(&w.output); err != nil {
+			return dst, err
+		}
+	}
+
+	defer func() {
+		w.output = *bytes.NewBuffer(nil)
+		w.writer.Reset(io.Discard)
+		c.writers.Put(w)
+	}()
+
+	if _, err := w.writer.Write(src); err != nil {
+		return w.output.Bytes(), err
+	}
+	if err := w.writer.Close(); err != nil {
+		return w.output.Bytes(), err
+	}
+	return w.output.Bytes(), nil
+}
+
+type Decompressor struct {
+	readers sync.Pool // *reader
+}
+
+type reader struct {
+	input  bytes.Reader
+	reader Reader
+}
+
+func (d *Decompressor) Decode(dst, src []byte, newReader func(io.Reader) (Reader, error)) ([]byte, error) {
+	r, _ := d.readers.Get().(*reader)
+	if r != nil {
+		r.input.Reset(src)
+		if err := r.reader.Reset(&r.input); err != nil {
+			return dst, err
+		}
+	} else {
+		r = new(reader)
+		r.input.Reset(src)
+		var err error
+		if r.reader, err = newReader(&r.input); err != nil {
+			return dst, err
+		}
+	}
+
+	defer func() {
+		r.input.Reset(nil)
+		if err := r.reader.Reset(nil); err == nil {
+			d.readers.Put(r)
+		}
+	}()
+
+	if cap(dst) == 0 {
+		dst = make([]byte, 0, 2*len(src))
+	} else {
+		dst = dst[:0]
+	}
+
+	for {
+		n, err := r.reader.Read(dst[len(dst):cap(dst)])
+		dst = dst[:len(dst)+n]
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			}
+			return dst, err
+		}
+
+		if len(dst) == cap(dst) {
+			tmp := make([]byte, len(dst), 2*len(dst))
+			copy(tmp, dst)
+			dst = tmp
+		}
+	}
 }
diff --git a/compress/gzip/gzip.go b/compress/gzip/gzip.go
index 7542025..b7375d1 100644
--- a/compress/gzip/gzip.go
+++ b/compress/gzip/gzip.go
@@ -24,6 +24,9 @@ const (
 
 type Codec struct {
 	Level int
+
+	r compress.Decompressor
+	w compress.Compressor
 }
 
 func (c *Codec) String() string {
@@ -34,43 +37,31 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Gzip
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	if r == nil {
-		r = strings.NewReader(emptyGzip)
-	}
-	z, err := gzip.NewReader(r)
-	if err != nil {
-		return nil, err
-	}
-	return reader{z}, nil
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	return c.w.Encode(dst, src, func(w io.Writer) (compress.Writer, error) {
+		return gzip.NewWriterLevel(w, c.Level)
+	})
 }
 
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	if w == nil {
-		w = io.Discard
-	}
-	z, err := gzip.NewWriterLevel(w, c.Level)
-	if err != nil {
-		return nil, err
-	}
-	return writer{z}, nil
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	return c.r.Decode(dst, src, func(r io.Reader) (compress.Reader, error) {
+		z, err := gzip.NewReader(r)
+		if err != nil {
+			return nil, err
+		}
+		return &reader{Reader: z}, nil
+	})
 }
 
-type reader struct{ *gzip.Reader }
+type reader struct {
+	*gzip.Reader
+	emptyGzip strings.Reader
+}
 
-func (r reader) Reset(rr io.Reader) error {
+func (r *reader) Reset(rr io.Reader) error {
 	if rr == nil {
-		rr = strings.NewReader(emptyGzip)
+		r.emptyGzip.Reset(emptyGzip)
+		rr = &r.emptyGzip
 	}
 	return r.Reader.Reset(rr)
 }
-
-type writer struct{ *gzip.Writer }
-
-func (w writer) Reset(ww io.Writer) error {
-	if ww == nil {
-		ww = io.Discard
-	}
-	w.Writer.Reset(ww)
-	return nil
-}
diff --git a/compress/lz4/lz4.go b/compress/lz4/lz4.go
index 18d97a9..4e096d4 100644
--- a/compress/lz4/lz4.go
+++ b/compress/lz4/lz4.go
@@ -2,11 +2,7 @@
 package lz4
 
 import (
-	"bytes"
-	"io"
-
 	"github.com/pierrec/lz4/v4"
-	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -41,68 +37,28 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Lz4Raw
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	return &reader{reader: r}, nil
-}
-
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	return &writer{writer: w, compressor: lz4.CompressorHC{Level: c.Level}}, nil
-}
-
-type reader struct {
-	buffer bytes.Buffer
-	data   []byte
-	offset int
-	reader io.Reader
-}
-
-func (r *reader) Close() error {
-	r.offset = len(r.data)
-	r.reader = nil
-	return nil
-}
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	dst = reserveAtLeast(dst, len(src)/4)
 
-func (r *reader) Reset(rr io.Reader) error {
-	r.buffer.Reset()
-	r.data = r.data[:0]
-	r.offset = 0
-	r.reader = rr
-	return nil
-}
-
-func (r *reader) Read(b []byte) (n int, err error) {
-	if r.offset == 0 && len(r.data) == 0 {
-		if err := r.decompress(); err != nil {
-			return 0, err
+	compressor := lz4.CompressorHC{Level: c.Level}
+	for {
+		n, err := compressor.CompressBlock(src, dst)
+		if err != nil { // see Decode for details about error handling
+			dst = make([]byte, 2*len(dst))
+		} else {
+			return dst[:n], nil
 		}
 	}
-	n = copy(b, r.data[r.offset:])
-	r.offset += n
-	if r.offset == len(r.data) {
-		err = io.EOF
-	}
-	return n, err
 }
 
-func (r *reader) decompress() error {
-	if r.reader == nil {
-		return io.EOF
-	}
-
-	_, err := r.buffer.ReadFrom(r.reader)
-	if err != nil {
-		return err
-	}
-
-	optimisticOutputSize := 3 * r.buffer.Len()
-	if cap(r.data) < optimisticOutputSize {
-		r.data = make([]byte, optimisticOutputSize)
-	} else {
-		r.data = r.data[:cap(r.data)]
-	}
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	// 3x seems like a common compression ratio, so we optimistically size the
+	// output buffer to that size. Feel free to change the value if you observe
+	// different behaviors.
+	dst = reserveAtLeast(dst, 3*len(src))
 
 	for {
-		n, err := lz4.UncompressBlock(r.buffer.Bytes(), r.data)
+		n, err := lz4.UncompressBlock(src, dst)
 		// The lz4 package does not expose the error values, they are declared
 		// in internal/lz4errors. Based on what I read of the implementation,
 		// the only condition where this function errors is if the output buffer
@@ -110,46 +66,18 @@ func (r *reader) decompress() error {
 		//
 		// https://github.com/pierrec/lz4/blob/a5532e5996ee86d17f8ce2694c08fb5bf3c6b471/internal/lz4block/block.go#L45-L53
 		if err != nil {
-			r.data = make([]byte, 2*len(r.data))
+			dst = make([]byte, 2*len(dst))
 		} else {
-			r.data = r.data[:n]
-			return nil
+			return dst[:n], nil
 		}
 	}
 }
 
-type writer struct {
-	buffer     bytes.Buffer
-	data       []byte
-	writer     io.Writer
-	compressor lz4.CompressorHC
-}
-
-func (w *writer) Reset(ww io.Writer) error {
-	w.buffer.Reset()
-	w.data = w.data[:0]
-	w.writer = ww
-	return nil
-}
-
-func (w *writer) Write(b []byte) (int, error) {
-	if w.writer == nil {
-		return 0, io.ErrClosedPipe
-	}
-	return w.buffer.Write(b)
-}
-
-func (w *writer) Close() (err error) {
-	if w.writer != nil && w.buffer.Len() > 0 {
-		limit := lz4.CompressBlockBound(w.buffer.Len())
-		if limit > cap(w.data) {
-			w.data = make([]byte, limit)
-		} else {
-			w.data = w.data[:limit]
-		}
-		size, _ := w.compressor.CompressBlock(w.buffer.Bytes(), w.data)
-		_, err = w.writer.Write(w.data[:size])
+func reserveAtLeast(b []byte, n int) []byte {
+	if cap(b) < n {
+		b = make([]byte, n)
+	} else {
+		b = b[:cap(b)]
 	}
-	w.writer = nil
-	return err
+	return b
 }
diff --git a/compress/snappy/snappy.go b/compress/snappy/snappy.go
index e7d796a..a6291bd 100644
--- a/compress/snappy/snappy.go
+++ b/compress/snappy/snappy.go
@@ -2,11 +2,7 @@
 package snappy
 
 import (
-	"bytes"
-	"io"
-
 	"github.com/klauspost/compress/snappy"
-	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -26,89 +22,10 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Snappy
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	return &reader{input: r, offset: -1}, nil
-}
-
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	return &writer{output: w}, nil
-}
-
-type reader struct {
-	input  io.Reader
-	buffer bytes.Buffer
-	offset int
-	data   []byte
-}
-
-func (r *reader) Close() error {
-	r.Reset(r.input)
-	return nil
-}
-
-func (r *reader) Reset(rr io.Reader) error {
-	r.input = rr
-	r.buffer.Reset()
-	r.offset = -1
-	r.data = r.data[:0]
-	return nil
-}
-
-func (r *reader) Read(b []byte) (int, error) {
-	if r.offset < 0 {
-		if r.input == nil {
-			return 0, io.EOF
-		}
-
-		_, err := r.buffer.ReadFrom(r.input)
-		if err != nil {
-			return 0, err
-		}
-
-		r.data, err = snappy.Decode(r.data[:0], r.buffer.Bytes())
-		if err != nil {
-			return 0, err
-		}
-
-		r.offset = 0
-	}
-
-	n := copy(b, r.data[r.offset:])
-	r.offset += n
-	if r.offset == len(r.data) {
-		return n, io.EOF
-	}
-	return n, nil
-}
-
-type writer struct {
-	output io.Writer
-	buffer []byte
-	data   []byte
-}
-
-func (w *writer) Close() error {
-	if w.output == nil {
-		w.buffer = w.buffer[:0]
-		return nil
-	}
-	if len(w.buffer) > 0 {
-		w.data = snappy.Encode(w.data[:0], w.buffer)
-		w.buffer = w.buffer[:0]
-	}
-	_, err := w.output.Write(w.data)
-	w.data = w.data[:0]
-	return err
-}
-
-func (w *writer) Reset(ww io.Writer) error {
-	w.output = ww
-	w.buffer = w.buffer[:0]
-	w.data = w.data[:0]
-	return nil
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	return snappy.Encode(dst, src), nil
 }
 
-func (w *writer) Write(b []byte) (int, error) {
-	w.buffer = append(w.buffer, b...)
-	return len(b), nil
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	return snappy.Decode(dst, src)
 }
diff --git a/compress/uncompressed/uncompressed.go b/compress/uncompressed/uncompressed.go
index e1c93f7..49da69c 100644
--- a/compress/uncompressed/uncompressed.go
+++ b/compress/uncompressed/uncompressed.go
@@ -4,9 +4,6 @@
 package uncompressed
 
 import (
-	"io"
-
-	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -21,20 +18,10 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Uncompressed
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	return &reader{r}, nil
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	return append(dst[:0], src...), nil
 }
 
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	return &writer{w}, nil
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	return append(dst[:0], src...), nil
 }
-
-type reader struct{ io.Reader }
-
-func (r *reader) Close() error             { return nil }
-func (r *reader) Reset(rr io.Reader) error { r.Reader = rr; return nil }
-
-type writer struct{ io.Writer }
-
-func (w *writer) Close() error             { return nil }
-func (w *writer) Reset(ww io.Writer) error { w.Writer = ww; return nil }
diff --git a/compress/zstd/zstd.go b/compress/zstd/zstd.go
index b1d8ccd..c26019e 100644
--- a/compress/zstd/zstd.go
+++ b/compress/zstd/zstd.go
@@ -2,10 +2,9 @@
 package zstd
 
 import (
-	"io"
+	"sync"
 
 	"github.com/klauspost/compress/zstd"
-	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -31,13 +30,14 @@ const (
 )
 
 const (
-	DefaultLevel       = SpeedDefault
-	DefaultConcurrency = 1
+	DefaultLevel = SpeedDefault
 )
 
 type Codec struct {
-	Level       Level
-	Concurrency int
+	Level Level
+
+	encoders sync.Pool // *zstd.Encoder
+	decoders sync.Pool // *zstd.Decoder
 }
 
 func (c *Codec) String() string {
@@ -48,34 +48,37 @@ func (c *Codec) CompressionCodec() format.CompressionCodec {
 	return format.Zstd
 }
 
-func (c *Codec) NewReader(r io.Reader) (compress.Reader, error) {
-	z, err := zstd.NewReader(r,
-		zstd.WithDecoderConcurrency(c.concurrency()),
-	)
-	if err != nil {
-		return nil, err
-	}
-	return reader{z}, nil
-}
-
-func (c *Codec) NewWriter(w io.Writer) (compress.Writer, error) {
-	z, err := zstd.NewWriter(nonNilWriter(w),
-		zstd.WithEncoderConcurrency(c.concurrency()),
-		zstd.WithEncoderLevel(c.level()),
-		zstd.WithZeroFrames(true),
-		zstd.WithEncoderCRC(false),
-	)
-	if err != nil {
-		return nil, err
+func (c *Codec) Encode(dst, src []byte) ([]byte, error) {
+	e, _ := c.encoders.Get().(*zstd.Encoder)
+	if e == nil {
+		var err error
+		e, err = zstd.NewWriter(nil,
+			zstd.WithEncoderConcurrency(1),
+			zstd.WithEncoderLevel(c.level()),
+			zstd.WithZeroFrames(true),
+			zstd.WithEncoderCRC(false),
+		)
+		if err != nil {
+			return dst[:0], err
+		}
 	}
-	return writer{z}, nil
+	defer c.encoders.Put(e)
+	return e.EncodeAll(src, dst[:0]), nil
 }
 
-func (c *Codec) concurrency() int {
-	if c.Concurrency != 0 {
-		return c.Concurrency
+func (c *Codec) Decode(dst, src []byte) ([]byte, error) {
+	d, _ := c.decoders.Get().(*zstd.Decoder)
+	if d == nil {
+		var err error
+		d, err = zstd.NewReader(nil,
+			zstd.WithDecoderConcurrency(1),
+		)
+		if err != nil {
+			return dst[:0], err
+		}
 	}
-	return DefaultConcurrency
+	defer c.decoders.Put(d)
+	return d.DecodeAll(src, dst[:0])
 }
 
 func (c *Codec) level() Level {
@@ -84,19 +87,3 @@ func (c *Codec) level() Level {
 	}
 	return DefaultLevel
 }
-
-type reader struct{ *zstd.Decoder }
-
-func (r reader) Close() error { r.Decoder.Close(); return nil }
-
-type writer struct{ *zstd.Encoder }
-
-func (w writer) Close() error             { w.Encoder.Close(); return nil }
-func (w writer) Reset(ww io.Writer) error { w.Encoder.Reset(nonNilWriter(ww)); return nil }
-
-func nonNilWriter(w io.Writer) io.Writer {
-	if w == nil {
-		w = io.Discard
-	}
-	return w
-}
diff --git a/file.go b/file.go
index 1cad2e5..702bf1c 100644
--- a/file.go
+++ b/file.go
@@ -11,7 +11,6 @@ import (
 	"sort"
 
 	"github.com/segmentio/encoding/thrift"
-	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -456,11 +455,6 @@ type filePages struct {
 	section *io.SectionReader
 	rbuf    *bufio.Reader
 
-	// This buffer holds compressed pages in memory when they are read; we need
-	// to read whole pages because we have to compute the checksum prior to
-	// exposing the page to the application.
-	compressedPageData []byte
-
 	page filePage
 	skip int64
 }
@@ -493,28 +487,24 @@ func (r *filePages) readPage() (*filePage, error) {
 		if err != io.EOF {
 			err = fmt.Errorf("decoding page header: %w", err)
 		}
-		if r.page.values != nil {
-			r.page.values.release()
-			r.page.values = nil
-		}
 		return nil, err
 	}
 
 	compressedPageSize := int(r.page.header.CompressedPageSize)
-	if cap(r.compressedPageData) < compressedPageSize {
-		r.compressedPageData = make([]byte, compressedPageSize)
+	if cap(r.page.data) < compressedPageSize {
+		r.page.data = make([]byte, compressedPageSize)
 	} else {
-		r.compressedPageData = r.compressedPageData[:compressedPageSize]
+		r.page.data = r.page.data[:compressedPageSize]
 	}
 
-	_, err := io.ReadFull(r.rbuf, r.compressedPageData)
+	_, err := io.ReadFull(r.rbuf, r.page.data)
 	if err != nil {
 		return nil, fmt.Errorf("reading page %d of column %q", r.page.index, r.page.columnPath())
 	}
 
 	if r.page.header.CRC != 0 {
 		headerChecksum := uint32(r.page.header.CRC)
-		bufferChecksum := crc32.ChecksumIEEE(r.compressedPageData)
+		bufferChecksum := crc32.ChecksumIEEE(r.page.data)
 
 		if headerChecksum != bufferChecksum {
 			// The parquet specs indicate that corruption errors could be
@@ -535,8 +525,6 @@ func (r *filePages) readPage() (*filePage, error) {
 		}
 	}
 
-	r.page.data.Reset(r.compressedPageData)
-
 	if r.column.columnIndex != nil {
 		err = r.page.parseColumnIndex(r.column.columnIndex)
 	} else {
@@ -558,18 +546,22 @@ func (r *filePages) readDictionary() error {
 }
 
 func (r *filePages) readDictionaryPage(p *filePage) error {
-	page := acquireCompressedPageReader(p.codec, &p.data)
+	pageData, err := p.decompress(p.data)
+	if err != nil {
+		return fmt.Errorf("decompressing dictionary page of column %q: %w", p.columnPath(), err)
+	}
+
 	enc := p.header.DictionaryPageHeader.Encoding
-	dec := LookupEncoding(enc).NewDecoder(page)
+	dec := LookupEncoding(enc).NewDecoder(bytes.NewReader(pageData))
 
 	columnIndex := r.column.Column()
 	numValues := int(p.NumValues())
 	dict, err := p.columnType.ReadDictionary(columnIndex, numValues, dec)
-	releaseCompressedPageReader(page)
 
 	if err != nil {
 		return fmt.Errorf("reading dictionary of column %q: %w", p.columnPath(), err)
 	}
+
 	r.page.dictionary = dict
 	r.page.columnType = dict.Type()
 	return nil
@@ -651,18 +643,13 @@ type filePage struct {
 
 	codec  format.CompressionCodec
 	header format.PageHeader
-	data   bytes.Reader
+	data   []byte
+	buffer []byte
 
 	index     int
 	minValue  Value
 	maxValue  Value
 	hasBounds bool
-
-	// This field caches the state used when reading values from the page.
-	// We allocate it separately to avoid creating it if the Values method
-	// is never called (e.g. when the page data is never decompressed).
-	// The state is released when the parent page reader reaches EOF.
-	values *filePageValueReaderState
 }
 
 var (
@@ -672,6 +659,18 @@ var (
 	errPageIndexExceedsColumnIndexNullCounts = errors.New("page index exceeds column index null counts")
 )
 
+func (p *filePage) decompress(pageData []byte) ([]byte, error) {
+	if p.codec != format.Uncompressed {
+		var err error
+		p.buffer, err = LookupCompressionCodec(p.codec).Decode(p.buffer[:0], pageData)
+		if err != nil {
+			return nil, err
+		}
+		pageData = p.buffer
+	}
+	return pageData, nil
+}
+
 func (p *filePage) statistics() *format.Statistics {
 	switch p.header.Type {
 	case format.DataPageV2:
@@ -829,118 +828,55 @@ func (p *filePage) Size() int64 {
 }
 
 func (p *filePage) Values() ValueReader {
-	if p.values == nil {
-		p.values = new(filePageValueReaderState)
-	}
-	if err := p.values.init(p.columnType, p.column, p.codec, p.PageHeader(), &p.data, p.dictionary); err != nil {
-		return &errorValueReader{err: err}
-	}
-	return p.values.reader
-}
-
-func (p *filePage) Buffer() BufferedPage {
-	bufferedPage := p.column.Type().NewColumnBuffer(p.Column(), int(p.Size()))
-	_, err := CopyValues(bufferedPage, p.Values())
+	v, err := p.values()
 	if err != nil {
-		return &errorPage{err: err, columnIndex: p.Column()}
+		v = &errorValueReader{err}
 	}
-	return bufferedPage.Page()
+	return v
 }
 
-func (p *filePage) PageHeader() PageHeader {
-	switch p.header.Type {
-	case format.DataPageV2:
-		return DataPageHeaderV2{p.header.DataPageHeaderV2}
-	case format.DataPage:
-		return DataPageHeaderV1{p.header.DataPageHeader}
-	case format.DictionaryPage:
-		return DictionaryPageHeader{p.header.DictionaryPageHeader}
-	default:
-		return unknownPageHeader{&p.header}
-	}
-}
-
-func (p *filePage) PageData() io.Reader { return &p.data }
-
-func (p *filePage) PageSize() int64 { return int64(p.header.CompressedPageSize) }
-
-func (p *filePage) CRC() uint32 { return uint32(p.header.CRC) }
-
-type filePageValueReaderState struct {
-	reader ColumnReader
-
-	v1 struct {
-		repetitions dataPageLevelV1
-		definitions dataPageLevelV1
-	}
-
-	v2 struct {
-		repetitions dataPageLevelV2
-		definitions dataPageLevelV2
-	}
-
-	repetitions struct {
-		encoding format.Encoding
-		decoder  encoding.Decoder
-	}
-
-	definitions struct {
-		encoding format.Encoding
-		decoder  encoding.Decoder
-	}
+func (p *filePage) values() (ValueReader, error) {
+	var repetitionLevels []byte
+	var definitionLevels []byte
+	var pageEncoding format.Encoding
+	var pageData = p.data
+	var numValues int
+	var err error
 
-	page struct {
-		encoding   format.Encoding
-		decoder    encoding.Decoder
-		compressed *compressedPageReader
-	}
-}
-
-func (s *filePageValueReaderState) release() {
-	if s.page.compressed != nil {
-		releaseCompressedPageReader(s.page.compressed)
-		s.page.compressed = nil
-	}
-}
-
-func (s *filePageValueReaderState) init(columnType Type, column *Column, codec format.CompressionCodec, header PageHeader, data *bytes.Reader, dict Dictionary) (err error) {
-	var repetitionLevels io.Reader
-	var definitionLevels io.Reader
-	var pageHeader DataPageHeader
-	var pageData io.Reader
+	maxRepetitionLevel := p.column.maxRepetitionLevel
+	maxDefinitionLevel := p.column.maxDefinitionLevel
 
-	switch h := header.(type) {
-	case DataPageHeaderV2:
-		repetitionLevels, definitionLevels, err = s.initDataPageV2(h, data)
+	switch p.header.Type {
+	case format.DataPageV2:
+		header := p.header.DataPageHeaderV2
+		repetitionLevels, definitionLevels, pageData, err = readDataPageV2(header, pageData)
 		if err != nil {
-			return fmt.Errorf("initializing v2 reader for page of column %q: %w", columnPath(column.Path()), err)
+			return nil, fmt.Errorf("initializing v2 reader for page of column %q: %w", p.columnPath(), err)
 		}
-		if h.IsCompressed(codec) {
-			s.page.compressed = makeCompressedPage(s.page.compressed, codec, data)
-			pageData = s.page.compressed
-		} else {
-			pageData = data
+		if p.codec != format.Uncompressed && (header.IsCompressed == nil || *header.IsCompressed) {
+			if pageData, err = p.decompress(pageData); err != nil {
+				return nil, fmt.Errorf("decompressing data page v2 of column %q: %w", p.columnPath(), err)
+			}
 		}
-		pageHeader = h
-
-	case DataPageHeaderV1:
-		if h.IsCompressed(codec) {
-			s.page.compressed = makeCompressedPage(s.page.compressed, codec, data)
-			pageData = s.page.compressed
-		} else {
-			pageData = data
+		pageEncoding = header.Encoding
+		numValues = int(header.NumValues)
+
+	case format.DataPage:
+		if pageData, err = p.decompress(pageData); err != nil {
+			return nil, fmt.Errorf("decompressing data page v1 of column %q: %w", p.columnPath(), err)
 		}
-		repetitionLevels, definitionLevels, err = s.initDataPageV1(column, pageData)
+		repetitionLevels, definitionLevels, pageData, err = readDataPageV1(maxRepetitionLevel, maxDefinitionLevel, pageData)
 		if err != nil {
-			return fmt.Errorf("initializing v1 reader for page of column %q: %w", columnPath(column.Path()), err)
+			return nil, fmt.Errorf("initializing v1 reader for page of column %q: %w", p.columnPath(), err)
 		}
-		pageHeader = h
+		header := p.header.DataPageHeader
+		pageEncoding = header.Encoding
+		numValues = int(header.NumValues)
 
 	default:
-		return fmt.Errorf("cannot read values of type %s from page of column %q", h.PageType(), columnPath(column.Path()))
+		return nil, fmt.Errorf("cannot read values of type %s from page of column %q", p.header.Type, p.columnPath())
 	}
 
-	pageEncoding := pageHeader.Encoding()
 	// In some legacy configurations, the PLAIN_DICTIONARY encoding is used on
 	// data page headers to indicate that the page contains indexes into the
 	// dictionary page, tho it is still encoded using the RLE encoding in this
@@ -949,164 +885,108 @@ func (s *filePageValueReaderState) init(columnType Type, column *Column, codec f
 	case format.PlainDictionary:
 		pageEncoding = format.RLEDictionary
 	}
-	s.page.decoder = makeDecoder(s.page.decoder, s.page.encoding, pageEncoding, pageData)
-	s.page.encoding = pageEncoding
 
-	maxRepetitionLevel := column.maxRepetitionLevel
-	maxDefinitionLevel := column.maxDefinitionLevel
+	pageDecoder := LookupEncoding(pageEncoding).NewDecoder(bytes.NewReader(pageData))
+	reader := p.columnType.NewColumnReader(int(p.column.index), defaultReadBufferSize)
+	reader.Reset(numValues, pageDecoder)
+
 	hasLevels := maxRepetitionLevel > 0 || maxDefinitionLevel > 0
 	if hasLevels {
-		repetitionLevelEncoding := pageHeader.RepetitionLevelEncoding()
-		definitionLevelEncoding := pageHeader.DefinitionLevelEncoding()
-		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.repetitions.encoding = repetitionLevelEncoding
-		s.definitions.encoding = definitionLevelEncoding
-	}
-
-	if s.reader == nil {
-		bufferSize := defaultReadBufferSize
-		if hasLevels {
-			bufferSize /= 2
-		}
-		s.reader = columnType.NewColumnReader(int(column.index), bufferSize)
-		if hasLevels {
-			s.reader = newFileColumnReader(s.reader, maxRepetitionLevel, maxDefinitionLevel, bufferSize)
-		}
-	}
-
-	numValues := int(header.NumValues())
-	switch r := s.reader.(type) {
-	case *fileColumnReader:
-		r.reset(numValues, s.repetitions.decoder, s.definitions.decoder, s.page.decoder)
-	default:
-		r.Reset(numValues, s.page.decoder)
+		repetitions := RLE.NewDecoder(bytes.NewReader(repetitionLevels))
+		definitions := RLE.NewDecoder(bytes.NewReader(definitionLevels))
+		fileReader := newFileColumnReader(reader, maxRepetitionLevel, maxDefinitionLevel, defaultReadBufferSize)
+		fileReader.reset(numValues, repetitions, definitions, pageDecoder)
+		reader = fileReader
 	}
 
-	return nil
+	return reader, nil
 }
 
-func (s *filePageValueReaderState) initDataPageV1(column *Column, data io.Reader) (repetitionLevels, definitionLevels io.Reader, err error) {
-	s.v1.repetitions.reset()
-	s.v1.definitions.reset()
-
-	if column.MaxRepetitionLevel() > 0 {
-		if err := s.v1.repetitions.readDataPageV1Level(data, "repetition"); err != nil {
-			return nil, nil, err
-		}
+func (p *filePage) Buffer() BufferedPage {
+	bufferedPage := p.column.Type().NewColumnBuffer(p.Column(), int(p.Size()))
+	_, err := CopyValues(bufferedPage, p.Values())
+	if err != nil {
+		return &errorPage{err: err, columnIndex: p.Column()}
 	}
+	return bufferedPage.Page()
+}
 
-	if column.MaxDefinitionLevel() > 0 {
-		if err := s.v1.definitions.readDataPageV1Level(data, "definition"); err != nil {
-			return nil, nil, err
-		}
+func (p *filePage) PageHeader() PageHeader {
+	switch p.header.Type {
+	case format.DataPageV2:
+		return DataPageHeaderV2{p.header.DataPageHeaderV2}
+	case format.DataPage:
+		return DataPageHeaderV1{p.header.DataPageHeader}
+	case format.DictionaryPage:
+		return DictionaryPageHeader{p.header.DictionaryPageHeader}
+	default:
+		return unknownPageHeader{&p.header}
 	}
-
-	repetitionLevels = &s.v1.repetitions.section
-	definitionLevels = &s.v1.definitions.section
-	return repetitionLevels, definitionLevels, nil
 }
 
-func (s *filePageValueReaderState) initDataPageV2(header DataPageHeaderV2, data *bytes.Reader) (repetitionLevels, definitionLevels io.Reader, err error) {
-	repetitionLevelsByteLength := header.RepetitionLevelsByteLength()
-	definitionLevelsByteLength := header.DefinitionLevelsByteLength()
+func (p *filePage) PageData() io.Reader { return bytes.NewReader(p.data) }
 
-	if repetitionLevelsByteLength > 0 {
-		s.v2.repetitions.setDataPageV2Section(data, 0, repetitionLevelsByteLength)
-	} else {
-		s.v2.repetitions.reset()
-	}
+func (p *filePage) PageSize() int64 { return int64(p.header.CompressedPageSize) }
 
-	if definitionLevelsByteLength > 0 {
-		s.v2.definitions.setDataPageV2Section(data, repetitionLevelsByteLength, definitionLevelsByteLength)
-	} else {
-		s.v2.definitions.reset()
-	}
+func (p *filePage) CRC() uint32 { return uint32(p.header.CRC) }
 
-	if _, err := data.Seek(repetitionLevelsByteLength+definitionLevelsByteLength, io.SeekStart); err != nil {
-		return nil, nil, err
+func readDataPageV1(maxRepetitionLevel, maxDefinitionLevel int8, page []byte) (repetitionLevels, definitionLevels, data []byte, err error) {
+	data = page
+	if maxRepetitionLevel > 0 {
+		repetitionLevels, data, err = readDataPageV1Level(data, "repetition")
+		if err != nil {
+			return nil, nil, page, err
+		}
 	}
-
-	repetitionLevels = &s.v2.repetitions.section
-	definitionLevels = &s.v2.definitions.section
-	return repetitionLevels, definitionLevels, nil
-}
-
-type dataPageLevelV1 struct {
-	section bytes.Reader
-	buffer  [4]byte
-	data    []byte
+	if maxDefinitionLevel > 0 {
+		definitionLevels, data, err = readDataPageV1Level(data, "definition")
+		if err != nil {
+			return nil, nil, page, err
+		}
+	}
+	return repetitionLevels, definitionLevels, data, nil
 }
 
-func (lvl *dataPageLevelV1) reset() {
-	lvl.section.Reset(nil)
+func readDataPageV1Level(page []byte, typ string) (level, data []byte, err error) {
+	size, page, err := read(page, 4)
+	if err != nil {
+		return nil, page, fmt.Errorf("reading %s level: %w", typ, err)
+	}
+	return read(page, int(binary.LittleEndian.Uint32(size)))
 }
 
-// This method breaks abstraction layers a bit, but it is helpful to avoid
-// decoding repetition and definition levels if there is no need to.
-//
-// In the data header format v1, the repetition and definition levels were
-// part of the compressed page payload. In order to access the data, the
-// levels must be fully read. Because of it, the levels have to be buffered
-// to allow the content to be decoded lazily later on.
-//
-// In the data header format v2, the repetition and definition levels are not
-// part of the compressed page data, they can be accessed by slicing a section
-// of the file according to the level lengths stored in the column metadata
-// header, therefore there is no need to buffer the levels.
-func (lvl *dataPageLevelV1) readDataPageV1Level(r io.Reader, typ string) error {
-	if _, err := io.ReadFull(r, lvl.buffer[:4]); err != nil {
-		return fmt.Errorf("reading RLE encoded length of %s levels: %w", typ, err)
-	}
-
-	n := int(binary.LittleEndian.Uint32(lvl.buffer[:4]))
-	if cap(lvl.data) < n {
-		lvl.data = make([]byte, n)
-	} else {
-		lvl.data = lvl.data[:n]
+func readDataPageV2(header *format.DataPageHeaderV2, page []byte) (repetitionLevels, definitionLevels, data []byte, err error) {
+	repetitionLevelsByteLength := header.RepetitionLevelsByteLength
+	definitionLevelsByteLength := header.DefinitionLevelsByteLength
+	data = page
+	if repetitionLevelsByteLength > 0 {
+		repetitionLevels, data, err = readDataPageV2Level(data, repetitionLevelsByteLength, "repetition")
+		if err != nil {
+			return nil, nil, page, err
+		}
 	}
-
-	if rn, err := io.ReadFull(r, lvl.data); err != nil {
-		return fmt.Errorf("reading %d/%d bytes of %s levels: %w", rn, n, typ, err)
+	if definitionLevelsByteLength > 0 {
+		definitionLevels, data, err = readDataPageV2Level(data, definitionLevelsByteLength, "definition")
+		if err != nil {
+			return nil, nil, page, err
+		}
 	}
-
-	lvl.section.Reset(lvl.data)
-	return nil
-}
-
-type dataPageLevelV2 struct {
-	section io.SectionReader
+	return repetitionLevels, definitionLevels, data, nil
 }
 
-func (lvl *dataPageLevelV2) reset() {
-	lvl.section = *io.NewSectionReader(nil, 0, 0)
-}
-
-func (lvl *dataPageLevelV2) setDataPageV2Section(file io.ReaderAt, dataPageOffset, dataPageLength int64) {
-	lvl.section = *io.NewSectionReader(file, dataPageOffset, dataPageLength)
-}
-
-func makeCompressedPage(page *compressedPageReader, codec format.CompressionCodec, compressed io.Reader) *compressedPageReader {
-	if page == nil {
-		page = acquireCompressedPageReader(codec, compressed)
-	} else {
-		if page.codec != codec {
-			releaseCompressedPageReader(page)
-			page = acquireCompressedPageReader(codec, compressed)
-		} else {
-			page.Reset(compressed)
-		}
+func readDataPageV2Level(page []byte, size int32, typ string) (level, data []byte, err error) {
+	level, data, err = read(page, int(size))
+	if err != nil {
+		err = fmt.Errorf("reading %s level: %w", typ, err)
 	}
-	return page
+	return level, data, err
 }
 
-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)
+func read(data []byte, size int) (head, tail []byte, err error) {
+	if len(data) < size {
+		return nil, data, io.ErrUnexpectedEOF
 	}
-	return decoder
+	return data[:size], data[size:], nil
 }
 
 var (
diff --git a/go.mod b/go.mod
index bac4f8e..c149187 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/andybalholm/brotli v1.0.3
 	github.com/google/uuid v1.3.0
 	github.com/hexops/gotextdiff v1.0.3
-	github.com/klauspost/compress v1.15.1
+	github.com/klauspost/compress v1.15.2
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pierrec/lz4/v4 v4.1.9
 	github.com/segmentio/encoding v0.3.5
diff --git a/go.sum b/go.sum
index ba451db..e5d4daa 100644
--- a/go.sum
+++ b/go.sum
@@ -4,8 +4,8 @@ github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=
-github.com/klauspost/compress v1.15.1 h1:y9FcTHGyrebwfP0ZZqFiaxTaiDnUrGkJkI+f583BL1A=
-github.com/klauspost/compress v1.15.1/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.15.2 h1:3WH+AG7s2+T8o3nrM/8u2rdqUEcQhmga7smjrT41nAw=
+github.com/klauspost/compress v1.15.2/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
diff --git a/page.go b/page.go
index 724b61d..f500d6a 100644
--- a/page.go
+++ b/page.go
@@ -56,9 +56,6 @@ type Page interface {
 
 	// Buffer returns the page as a BufferedPage, which may be the page itself
 	// if it was already buffered.
-	//
-	// Compressed pages will be consumed to create the returned buffered page,
-	// their content will no be readable anymore after the call.
 	Buffer() BufferedPage
 }
 
diff --git a/page_header.go b/page_header.go
index 7f21881..73646ca 100644
--- a/page_header.go
+++ b/page_header.go
@@ -23,10 +23,6 @@ type PageHeader interface {
 type DataPageHeader interface {
 	PageHeader
 
-	// Returns whether the page is compressed, according to the codec given as
-	// argument and details stored in the page header.
-	IsCompressed(format.CompressionCodec) bool
-
 	// Returns the encoding of the repetition level section.
 	RepetitionLevelEncoding() format.Encoding
 
@@ -96,10 +92,6 @@ func (v1 DataPageHeaderV1) NumValues() int64 {
 	return int64(v1.header.NumValues)
 }
 
-func (v1 DataPageHeaderV1) IsCompressed(codec format.CompressionCodec) bool {
-	return codec != format.Uncompressed
-}
-
 func (v1 DataPageHeaderV1) RepetitionLevelEncoding() format.Encoding {
 	return v1.header.RepetitionLevelEncoding
 }
@@ -152,10 +144,6 @@ func (v2 DataPageHeaderV2) NumRows() int64 {
 	return int64(v2.header.NumRows)
 }
 
-func (v2 DataPageHeaderV2) IsCompressed(codec format.CompressionCodec) bool {
-	return codec != format.Uncompressed && (v2.header.IsCompressed == nil || *v2.header.IsCompressed)
-}
-
 func (v2 DataPageHeaderV2) RepetitionLevelsByteLength() int64 {
 	return int64(v2.header.RepetitionLevelsByteLength)
 }
diff --git a/reader.go b/reader.go
index 208f086..1809441 100644
--- a/reader.go
+++ b/reader.go
@@ -179,7 +179,7 @@ func (r *Reader) Read(row interface{}) (err error) {
 	}
 
 	if err := r.read.SeekToRow(r.rowIndex); err != nil {
-		return err
+		return fmt.Errorf("seeking reader to row %d: %w", r.rowIndex, err)
 	}
 
 	r.values, err = r.read.ReadRow(r.values[:0])
diff --git a/writer.go b/writer.go
index 4525b24..60673be 100644
--- a/writer.go
+++ b/writer.go
@@ -201,11 +201,6 @@ type writer struct {
 	createdBy string
 	metadata  []format.KeyValue
 
-	buffers struct {
-		header bytes.Buffer
-		page   bytes.Buffer
-	}
-
 	columns       []*writerColumn
 	columnChunk   []format.ColumnChunk
 	columnIndex   []format.ColumnIndex
@@ -220,6 +215,13 @@ type writer struct {
 	sortingColumns []format.SortingColumn
 }
 
+type writerBuffers struct {
+	compressed []byte
+	header     bytes.Buffer
+	page       bytes.Buffer
+	reader     bytes.Reader
+}
+
 func newWriter(output io.Writer, config *WriterConfig) *writer {
 	w := new(writer)
 	if config.WriteBufferSize <= 0 {
@@ -282,6 +284,12 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		defaultCompression = &Uncompressed
 	}
 
+	// Those buffers are scratch space used to generate the page header and
+	// content, they are shared by all column chunks because they are only
+	// used during calls to writeDictionaryPage or writeDataPage, which are
+	// not done concurrently.
+	buffers := new(writerBuffers)
+
 	forEachLeafColumnOf(config.Schema, func(leaf leafColumn) {
 		encoding := encodingOf(leaf.node)
 		dictionary := Dictionary(nil)
@@ -299,6 +307,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		}
 
 		c := &writerColumn{
+			buffers:            buffers,
 			pool:               config.ColumnPageBuffers,
 			columnPath:         leaf.path,
 			columnType:         columnType,
@@ -321,12 +330,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 			isCompressed: compression.CompressionCodec() != format.Uncompressed && (dataPageType != format.DataPageV2 || dictionary == nil),
 		}
 
-		// Those buffers are scratch space used to generate the page header and
-		// content, they are shared by all column chunks because they are only
-		// used during calls to writeDictionaryPage or writeDataPage, which are
-		// not done concurrently.
-		c.header.buffer, c.page.buffer = &w.buffers.header, &w.buffers.page
-		c.header.encoder.Reset(c.header.protocol.NewWriter(c.header.buffer))
+		c.header.encoder.Reset(c.header.protocol.NewWriter(&buffers.header))
 
 		if leaf.maxRepetitionLevel > 0 {
 			c.insert = (*writerColumn).insertRepeated
@@ -681,32 +685,21 @@ type writerColumn struct {
 	maxRepetitionLevel int8
 	maxDefinitionLevel int8
 
+	buffers *writerBuffers
+
 	levels struct {
 		encoder encoding.Encoder
-		// In data pages v1, the repetition and definition levels are prefixed
-		// with the 4 bytes length of the sections. While the parquet-format
-		// documentation indicates that the length prefix is part of the hybrid
-		// RLE/Bit-Pack encoding, this is the only condition where it is used
-		// so we treat it as a special case rather than implementing it in the
-		// encoding.
-		//
-		// Reference https://github.com/apache/parquet-format/blob/master/Encodings.md#run-length-encoding--bit-packing-hybrid-rle--3
-		v1 lengthPrefixedWriter
 	}
 
 	header struct {
-		buffer   *bytes.Buffer
 		protocol thrift.CompactProtocol
 		encoder  thrift.Encoder
 	}
 
 	page struct {
-		buffer       *bytes.Buffer
-		filter       *bloomFilterEncoder
-		compressed   compress.Writer
-		uncompressed offsetTrackingWriter
-		encoding     format.Encoding
-		encoder      encoding.Encoder
+		filter   *bloomFilterEncoder
+		encoding format.Encoding
+		encoder  encoding.Encoder
 	}
 
 	dict struct {
@@ -958,51 +951,51 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 		return 0, nil
 	}
 
-	c.page.buffer.Reset()
-	repetitionLevelsByteLength := int32(0)
-	definitionLevelsByteLength := int32(0)
+	buffer := &c.buffers.page
+	buffer.Reset()
+	repetitionLevelsByteLength := 0
+	definitionLevelsByteLength := 0
 
-	if c.dataPageType == format.DataPageV2 {
+	switch c.dataPageType {
+	case format.DataPageV2:
 		if c.maxRepetitionLevel > 0 {
-			c.page.uncompressed.Reset(c.page.buffer)
-			c.levels.encoder.Reset(&c.page.uncompressed)
+			c.levels.encoder.Reset(buffer)
 			c.levels.encoder.SetBitWidth(bits.Len8(c.maxRepetitionLevel))
 			c.levels.encoder.EncodeInt8(page.RepetitionLevels())
-			repetitionLevelsByteLength = int32(c.page.uncompressed.offset)
+			repetitionLevelsByteLength = buffer.Len()
 		}
 		if c.maxDefinitionLevel > 0 {
-			c.page.uncompressed.Reset(c.page.buffer)
-			c.levels.encoder.Reset(&c.page.uncompressed)
+			c.levels.encoder.Reset(buffer)
 			c.levels.encoder.SetBitWidth(bits.Len8(c.maxDefinitionLevel))
 			c.levels.encoder.EncodeInt8(page.DefinitionLevels())
-			definitionLevelsByteLength = int32(c.page.uncompressed.offset)
+			definitionLevelsByteLength = buffer.Len() - repetitionLevelsByteLength
 		}
-	}
 
-	if !c.isCompressed {
-		c.page.uncompressed.Reset(c.page.buffer)
-	} else {
-		p, err := c.compressedPage(c.page.buffer)
-		if err != nil {
-			return 0, err
-		}
-		c.page.uncompressed.Reset(p)
-	}
-
-	if c.dataPageType == format.DataPage {
+	case format.DataPage:
+		// In data pages v1, the repetition and definition levels are prefixed
+		// with the 4 bytes length of the sections. While the parquet-format
+		// documentation indicates that the length prefix is part of the hybrid
+		// RLE/Bit-Pack encoding, this is the only condition where it is used
+		// so we treat it as a special case rather than implementing it in the
+		// encoding.
+		//
+		// Reference https://github.com/apache/parquet-format/blob/master/Encodings.md#run-length-encoding--bit-packing-hybrid-rle--3
+		lengthPlaceholder := make([]byte, 4)
 		if c.maxRepetitionLevel > 0 {
-			c.levels.v1.Reset(&c.page.uncompressed)
-			c.levels.encoder.Reset(&c.levels.v1)
+			buffer.Write(lengthPlaceholder)
+			offset := buffer.Len()
+			c.levels.encoder.Reset(buffer)
 			c.levels.encoder.SetBitWidth(bits.Len8(c.maxRepetitionLevel))
 			c.levels.encoder.EncodeInt8(page.RepetitionLevels())
-			c.levels.v1.Close()
+			binary.LittleEndian.PutUint32(buffer.Bytes()[offset-4:], uint32(buffer.Len()-offset))
 		}
 		if c.maxDefinitionLevel > 0 {
-			c.levels.v1.Reset(&c.page.uncompressed)
-			c.levels.encoder.Reset(&c.levels.v1)
+			buffer.Write(lengthPlaceholder)
+			offset := buffer.Len()
+			c.levels.encoder.Reset(buffer)
 			c.levels.encoder.SetBitWidth(bits.Len8(c.maxDefinitionLevel))
 			c.levels.encoder.EncodeInt8(page.DefinitionLevels())
-			c.levels.v1.Close()
+			binary.LittleEndian.PutUint32(buffer.Bytes()[offset-4:], uint32(buffer.Len()-offset))
 		}
 	}
 
@@ -1020,26 +1013,34 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 		statistics = c.makePageStatistics(page)
 	}
 
-	c.page.encoder.Reset(&c.page.uncompressed)
+	c.page.encoder.Reset(buffer)
 	if err := page.WriteTo(c.page.encoder); err != nil {
 		return 0, err
 	}
-	if c.page.compressed != nil {
-		if err := c.page.compressed.Close(); err != nil {
-			return 0, err
+
+	uncompressedPageSize := buffer.Len()
+	pageData := buffer.Bytes()
+	if c.isCompressed {
+		offset := repetitionLevelsByteLength + definitionLevelsByteLength
+		b, err := c.compress(pageData[offset:])
+		if err != nil {
+			return 0, fmt.Errorf("compressing parquet data page: %w", err)
+		}
+		if offset == 0 {
+			pageData = b
+		} else {
+			// TODO: can this copy be optimized away?
+			buffer.Truncate(offset)
+			buffer.Write(b)
+			pageData = buffer.Bytes()
 		}
 	}
 
-	c.header.buffer.Reset()
-	levelsByteLength := repetitionLevelsByteLength + definitionLevelsByteLength
-	uncompressedPageSize := c.page.uncompressed.offset + int64(levelsByteLength)
-	compressedPageSize := c.page.buffer.Len()
-
 	pageHeader := &format.PageHeader{
 		Type:                 c.dataPageType,
 		UncompressedPageSize: int32(uncompressedPageSize),
-		CompressedPageSize:   int32(compressedPageSize),
-		CRC:                  int32(crc32.ChecksumIEEE(c.page.buffer.Bytes())),
+		CompressedPageSize:   int32(len(pageData)),
+		CRC:                  int32(crc32.ChecksumIEEE(pageData)),
 	}
 
 	numRows := page.NumRows()
@@ -1059,21 +1060,28 @@ func (c *writerColumn) writeBufferedPage(page BufferedPage) (int64, error) {
 			NumNulls:                   int32(numNulls),
 			NumRows:                    int32(numRows),
 			Encoding:                   c.page.encoding,
-			DefinitionLevelsByteLength: definitionLevelsByteLength,
-			RepetitionLevelsByteLength: repetitionLevelsByteLength,
+			DefinitionLevelsByteLength: int32(definitionLevelsByteLength),
+			RepetitionLevelsByteLength: int32(repetitionLevelsByteLength),
 			IsCompressed:               &c.isCompressed,
 			Statistics:                 statistics,
 		}
 	}
 
+	header := &c.buffers.header
+	header.Reset()
 	if err := c.header.encoder.Encode(pageHeader); err != nil {
 		return 0, err
 	}
-	headerSize := int32(c.header.buffer.Len())
-	compressedSize := int64(headerSize) + int64(compressedPageSize)
-	if err := c.writePage(compressedSize, c.header.buffer, c.page.buffer); err != nil {
+	headerSize := int32(header.Len())
+	compressedSize := int64(headerSize) + int64(len(pageData))
+
+	reader := &c.buffers.reader
+	reader.Reset(pageData)
+
+	if err := c.writePage(compressedSize, header, reader); err != nil {
 		return 0, err
 	}
+
 	c.recordPageStats(headerSize, pageHeader, page)
 	return numValues, nil
 }
@@ -1115,15 +1123,18 @@ func (c *writerColumn) writeCompressedPage(page CompressedPage) (int64, error) {
 		return 0, fmt.Errorf("writing compressed page type of unknown type: %s", h.PageType())
 	}
 
-	c.header.buffer.Reset()
+	header := &c.buffers.header
+	header.Reset()
 	if err := c.header.encoder.Encode(pageHeader); err != nil {
 		return 0, err
 	}
-	headerSize := int32(c.header.buffer.Len())
+	headerSize := int32(header.Len())
 	compressedSize := int64(headerSize + pageHeader.CompressedPageSize)
-	if err := c.writePage(compressedSize, c.header.buffer, page.PageData()); err != nil {
+
+	if err := c.writePage(compressedSize, header, page.PageData()); err != nil {
 		return 0, err
 	}
+
 	c.recordPageStats(headerSize, pageHeader, page)
 	return page.NumValues(), nil
 }
@@ -1153,28 +1164,25 @@ func (c *writerColumn) writePage(size int64, header, data io.Reader) error {
 }
 
 func (c *writerColumn) writeDictionaryPage(output io.Writer, dict Dictionary) error {
-	c.page.buffer.Reset()
-
-	p, err := c.compressedPage(c.page.buffer)
-	if err != nil {
-		return err
-	}
-
-	c.page.uncompressed.Reset(p)
-	c.dict.encoder.Reset(&c.page.uncompressed)
+	buffer := &c.buffers.page
+	buffer.Reset()
+	c.dict.encoder.Reset(buffer)
 
 	if err := dict.Page().WriteTo(&c.dict.encoder); err != nil {
 		return fmt.Errorf("writing parquet dictionary page: %w", err)
 	}
-	if err := p.Close(); err != nil {
-		return fmt.Errorf("flushing compressed parquet dictionary page: %w", err)
+
+	uncompressedPageSize := buffer.Len()
+	pageData, err := c.compress(buffer.Bytes())
+	if err != nil {
+		return fmt.Errorf("compressing parquet dictionary page: %w", err)
 	}
 
 	pageHeader := &format.PageHeader{
 		Type:                 format.DictionaryPage,
-		UncompressedPageSize: int32(c.page.uncompressed.offset),
-		CompressedPageSize:   int32(c.page.buffer.Len()),
-		CRC:                  int32(crc32.ChecksumIEEE(c.page.buffer.Bytes())),
+		UncompressedPageSize: int32(uncompressedPageSize),
+		CompressedPageSize:   int32(len(pageData)),
+		CRC:                  int32(crc32.ChecksumIEEE(pageData)),
 		DictionaryPageHeader: &format.DictionaryPageHeader{
 			NumValues: int32(dict.Len()),
 			Encoding:  format.Plain,
@@ -1182,33 +1190,31 @@ func (c *writerColumn) writeDictionaryPage(output io.Writer, dict Dictionary) er
 		},
 	}
 
-	c.header.buffer.Reset()
+	header := &c.buffers.header
+	header.Reset()
 	if err := c.header.encoder.Encode(pageHeader); err != nil {
 		return err
 	}
-	if _, err := output.Write(c.header.buffer.Bytes()); err != nil {
+	if _, err := output.Write(header.Bytes()); err != nil {
 		return err
 	}
-	if _, err := output.Write(c.page.buffer.Bytes()); err != nil {
+	if _, err := output.Write(pageData); err != nil {
 		return err
 	}
-	c.recordPageStats(int32(c.header.buffer.Len()), pageHeader, nil)
+	c.recordPageStats(int32(header.Len()), pageHeader, nil)
 	return nil
 }
 
-func (c *writerColumn) compressedPage(w io.Writer) (compress.Writer, error) {
-	if c.page.compressed == nil {
-		z, err := c.compression.NewWriter(w)
+func (c *writerColumn) compress(pageData []byte) ([]byte, error) {
+	if c.compression.CompressionCodec() != format.Uncompressed {
+		b, err := c.compression.Encode(c.buffers.compressed[:0], pageData)
+		c.buffers.compressed = b
 		if err != nil {
-			return nil, fmt.Errorf("creating compressor for parquet column chunk writer: %w", err)
-		}
-		c.page.compressed = z
-	} else {
-		if err := c.page.compressed.Reset(w); err != nil {
-			return nil, fmt.Errorf("resetting compressor for parquet column chunk writer: %w", err)
+			return nil, err
 		}
+		pageData = b
 	}
-	return c.page.compressed, nil
+	return pageData, nil
 }
 
 func (c *writerColumn) makePageStatistics(page Page) format.Statistics {
@@ -1305,31 +1311,6 @@ func sortPageEncodingStats(stats []format.PageEncodingStats) {
 	})
 }
 
-type lengthPrefixedWriter struct {
-	writer io.Writer
-	buffer []byte
-}
-
-func (w *lengthPrefixedWriter) Reset(ww io.Writer) {
-	w.writer = ww
-	w.buffer = append(w.buffer[:0], 0, 0, 0, 0)
-}
-
-func (w *lengthPrefixedWriter) Close() error {
-	if len(w.buffer) > 0 {
-		defer func() { w.buffer = w.buffer[:0] }()
-		binary.LittleEndian.PutUint32(w.buffer, uint32(len(w.buffer))-4)
-		_, err := w.writer.Write(w.buffer)
-		return err
-	}
-	return nil
-}
-
-func (w *lengthPrefixedWriter) Write(b []byte) (int, error) {
-	w.buffer = append(w.buffer, b...)
-	return len(b), nil
-}
-
 type offsetTrackingWriter struct {
 	writer io.Writer
 	offset int64


Test patch:
diff --git a/compress/compress_test.go b/compress/compress_test.go
index 2eea73f..5fae54d 100644
--- a/compress/compress_test.go
+++ b/compress/compress_test.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"io"
 	"testing"
-	"testing/iotest"
 
 	"github.com/segmentio/parquet-go/compress"
 	"github.com/segmentio/parquet-go/compress/brotli"
@@ -15,91 +14,137 @@ import (
 	"github.com/segmentio/parquet-go/compress/zstd"
 )
 
-func TestCompressionCodec(t *testing.T) {
-	tests := []struct {
-		scenario string
-		codec    compress.Codec
-	}{
-		{
-			scenario: "uncompressed",
-			codec:    new(uncompressed.Codec),
-		},
-
-		{
-			scenario: "snappy",
-			codec:    new(snappy.Codec),
-		},
-
-		{
-			scenario: "gzip",
-			codec:    new(gzip.Codec),
-		},
-
-		{
-			scenario: "brotli",
-			codec:    new(brotli.Codec),
-		},
-
-		{
-			scenario: "zstd",
-			codec:    new(zstd.Codec),
-		},
-
-		{
-			scenario: "lz4",
-			codec:    new(lz4.Codec),
-		},
-	}
+var tests = [...]struct {
+	scenario string
+	codec    compress.Codec
+}{
+	{
+		scenario: "uncompressed",
+		codec:    new(uncompressed.Codec),
+	},
 
-	buffer := new(bytes.Buffer)
-	output := new(bytes.Buffer)
-	random := bytes.Repeat([]byte("1234567890qwertyuiopasdfghjklzxcvbnm"), 1000)
+	{
+		scenario: "snappy",
+		codec:    new(snappy.Codec),
+	},
 
-	for _, test := range tests {
-		t.Run(test.scenario, func(t *testing.T) {
-			w, err := test.codec.NewWriter(nil)
-			if err != nil {
-				t.Fatal(err)
-			}
-			defer w.Close()
+	{
+		scenario: "gzip",
+		codec:    new(gzip.Codec),
+	},
 
-			r, err := test.codec.NewReader(nil)
-			if err != nil {
-				t.Fatal(err)
-			}
-			defer r.Close()
+	{
+		scenario: "brotli",
+		codec:    new(brotli.Codec),
+	},
 
-			for i := 0; i < 10; i++ {
-				buffer.Reset()
-				output.Reset()
+	{
+		scenario: "zstd",
+		codec:    new(zstd.Codec),
+	},
 
-				if err := w.Reset(buffer); err != nil {
-					t.Fatal(err)
-				}
-				if _, err := io.Copy(w, iotest.OneByteReader(bytes.NewReader(random))); err != nil {
-					t.Fatal(err)
-				}
-				if err := w.Close(); err != nil {
-					t.Fatal(err)
-				}
+	{
+		scenario: "lz4",
+		codec:    new(lz4.Codec),
+	},
+}
 
-				if err := r.Reset(buffer); err != nil {
-					t.Fatal(err)
-				}
-				if _, err := io.Copy(output, iotest.OneByteReader(r)); err != nil {
+var testdata = bytes.Repeat([]byte("1234567890qwertyuiopasdfghjklzxcvbnm"), 10e3)
+
+func TestCompressionCodec(t *testing.T) {
+	buffer := make([]byte, 0, len(testdata))
+	output := make([]byte, 0, len(testdata))
+
+	for _, test := range tests {
+		t.Run(test.scenario, func(t *testing.T) {
+			const N = 10
+			// Run the test multiple times to exercise codecs that maintain
+			// state across compression/decompression.
+			for i := 0; i < N; i++ {
+				var err error
+
+				buffer, err = test.codec.Encode(buffer[:0], testdata)
+				if err != nil {
 					t.Fatal(err)
 				}
-				if !bytes.Equal(random, output.Bytes()) {
-					t.Errorf("content mismatch after compressing and decompressing:\n%q\n%q", random, output.Bytes())
-				}
 
-				if err := w.Reset(nil); err != nil {
+				output, err = test.codec.Decode(output[:0], buffer)
+				if err != nil {
 					t.Fatal(err)
 				}
-				if err := r.Reset(nil); err != nil {
-					t.Fatal(err)
+
+				if !bytes.Equal(testdata, output) {
+					t.Errorf("content mismatch after compressing and decompressing (attempt %d/%d)", i+1, N)
 				}
 			}
 		})
 	}
 }
+
+func BenchmarkEncode(b *testing.B) {
+	buffer := make([]byte, 0, len(testdata))
+
+	for _, test := range tests {
+		b.Run(test.scenario, func(b *testing.B) {
+			b.SetBytes(int64(len(testdata)))
+			benchmarkZeroAllocsPerRun(b, func() {
+				buffer, _ = test.codec.Encode(buffer[:0], testdata)
+			})
+		})
+	}
+}
+
+func BenchmarkDecode(b *testing.B) {
+	buffer := make([]byte, 0, len(testdata))
+	output := make([]byte, 0, len(testdata))
+
+	for _, test := range tests {
+		b.Run(test.scenario, func(b *testing.B) {
+			buffer, _ = test.codec.Encode(buffer[:0], testdata)
+			b.SetBytes(int64(len(testdata)))
+			benchmarkZeroAllocsPerRun(b, func() {
+				output, _ = test.codec.Encode(output[:0], buffer)
+			})
+		})
+	}
+}
+
+type simpleReader struct{ io.Reader }
+
+func (s *simpleReader) Close() error            { return nil }
+func (s *simpleReader) Reset(r io.Reader) error { s.Reader = r; return nil }
+
+type simpleWriter struct{ io.Writer }
+
+func (s *simpleWriter) Close() error      { return nil }
+func (s *simpleWriter) Reset(w io.Writer) { s.Writer = w }
+
+func BenchmarkCompressor(b *testing.B) {
+	compressor := compress.Compressor{}
+	src := make([]byte, 1000)
+	dst := make([]byte, 1000)
+
+	benchmarkZeroAllocsPerRun(b, func() {
+		dst, _ = compressor.Encode(dst, src, func(w io.Writer) (compress.Writer, error) {
+			return &simpleWriter{Writer: w}, nil
+		})
+	})
+}
+
+func BenchmarkDecompressor(b *testing.B) {
+	decompressor := compress.Decompressor{}
+	src := make([]byte, 1000)
+	dst := make([]byte, 1000)
+
+	benchmarkZeroAllocsPerRun(b, func() {
+		dst, _ = decompressor.Decode(dst, src, func(r io.Reader) (compress.Reader, error) {
+			return &simpleReader{Reader: r}, nil
+		})
+	})
+}
+
+func benchmarkZeroAllocsPerRun(b *testing.B, f func()) {
+	if allocs := testing.AllocsPerRun(b.N, f); allocs != 0 {
+		b.Errorf("too many memory allocations: %g > 0", allocs)
+	}
+}
diff --git a/reader_test.go b/reader_test.go
index 3ac0743..2e8ba97 100644
--- a/reader_test.go
+++ b/reader_test.go
@@ -65,14 +65,14 @@ type decimalColumn struct {
 }
 
 type addressBook struct {
-	Owner             utf8string
-	OwnerPhoneNumbers []utf8string
+	Owner             utf8string   `parquet:",plain"`
+	OwnerPhoneNumbers []utf8string `parquet:",plain"`
 	Contacts          []contact
 }
 
 type contact struct {
-	Name        utf8string
-	PhoneNumber utf8string
+	Name        utf8string `parquet:",plain"`
+	PhoneNumber utf8string `parquet:",plain"`
 }
 
 type listColumn2 struct {