Repository: segmentio/parquet-go
Pull request #72 (2022-02-18 03:05:25)

[O11Y-218] add parquet.ColumnReader API to allow reading columns of typed values from parquet files
This PR is a follow up to #64, it modifies the internal implementation of reading file pages to expose types that implement the `parquet.[Type]Reader` interfaces as a mechanism to read columns of typed values from parquet files (prior to this change, it was only possible to do so from in-memory buffers).

I still need to add documentations describing how to leverage those interfaces, but the code changes should be ready to review.


Patch:
diff --git a/README.md b/README.md
index a80e3de..36e449a 100644
--- a/README.md
+++ b/README.md
@@ -39,6 +39,10 @@ Go 1.17 or later is required to use the package.
 
 ## Usage
 
+The following sections describe how to use APIs exposed by the library,
+highlighting the use cases with code examples to demonstrate how they are used
+in practice.
+
 ### Writing Parquet Files: [parquet.Writer](https://pkg.go.dev/github.com/segmentio/parquet-go#Writer)
 
 A parquet file is a collection of rows sharing the same schema, arranged in
@@ -353,6 +357,130 @@ for i, n := 0, file.NumRowGroups(); i < n; i++ {
 }
 ```
 
+## Optimizations
+
+The following sections describe common optimization techniques supported by the
+library.
+
+### Optimizing Reads
+
+Lower level APIs used to read parquet files offer more efficient ways to access
+column values. Consecutive sequences of values are grouped into pages which are
+represented by the `parquet.Page` interface.
+
+A column chunk may contain multiple pages, each holding a section of the column
+values. Applications can retrieve the column values either by reading them into
+buffers of `parquet.Value`, or type asserting the pages to read arrays of
+primitive Go values. The following example demonstrates how to use both
+mechanisms to read column values:
+
+```go
+pages := column.Pages()
+
+for {
+    p, err := pages.ReadPage()
+    if err != nil {
+        ... // io.EOF when there are no more pages
+    }
+
+    switch page := p.Values().(type) {
+    case parquet.Int32Page:
+        values := make([]int32, page.NumValues())
+        _, err := page.ReadInt32s(values)
+        ...
+    case parquet.Int64Page:
+        values := make([]int64, page.NumValues())
+        _, err := page.ReadInt64s(values)
+        ...
+    default:
+        values := make([]parquet.Value, page.NumValues())
+        _, err := page.ReadValues(values)
+        ...
+    }
+}
+```
+
+Reading arrays of typed values is often preferable when performing aggregations
+on the values as this model offers a more compact representation of the values
+in memory, and pairs well with the use of optimizations like SIMD vectorization.
+
+### Optimizing Writes
+
+Applications that deal with columnar storage are sometimes designed to work with
+columnar data throughout the abstraction layers; it then becomes possible to
+write columns of values directly instead of reconstructing rows from the column
+values. The package offers two main mechanisms to satisfy those use cases:
+
+#### A. Writing Columns of Typed Arrays
+
+The first solution assumes that the program works with in-memory arrays of typed
+values, for example slices of primitive Go types like `[]float32`; this would be
+the case if the application is built on top of a framework like
+[Apache Arrow](https://pkg.go.dev/github.com/apache/arrow/go/arrow).
+
+`parquet.Buffer` is an implementation of the `parquet.RowGroup` interface which
+maintains in-memory buffers of column values. Rows can be written by either
+boxing primitive values into arrays of `parquet.Value`, or type asserting the
+columns to a access specialized versions of the write methods accepting arrays
+of Go primitive types.
+
+When using either of these models, the application is responsible for ensuring
+that the same number of rows are written to each column or the resulting parquet
+file will be malformed.
+
+The following examples demonstrate how to use these two models to write columns
+of Go values:
+
+```go
+func writeColumns(buffer *parquet.Buffer, columns [3][]interface{}) error {
+    values := make([]parquet.Value, len(columns[0]))
+    for i := range columns {
+        c := buffer.Column(i).(parquet.ColumnBuffer)
+        for j, v := range columns[i] {
+            values[j] = parquet.ValueOf(v)
+        }
+        if _, err := c.WriteValues(values); err != nil {
+            return err
+        }
+    }
+    return nil
+}
+```
+
+```go
+func writeColumns(buffer *parquet.Buffer, ids []int64, values []float32) error {
+    if len(ids) != len(values) {
+        return fmt.Errorf("number of ids and values mismatch: ids=%d values=%d", len(ids), len(values))
+    }
+    if err := buffer.(parquet.Int64Writer).WriteInt64s(ids); err != nil {
+        return err
+    }
+    if err := buffer.(parquet.FloatWriter).WriteFloats(values); err != nil {
+        return err
+    }
+    return nil
+}
+```
+
+The latter is more efficient as it does not require boxing the input into an
+intermediary array of `parquet.Value`. However, it may not always be the right
+model depending on the situation, sometimes the generic abstraction can be a
+more expressive model.
+
+#### B. Implementing parquet.RowGroup
+
+Programs that need full control over the construction of row groups can choose
+to provide their own implementation of the `parquet.RowGroup` interface, which
+includes defining implementations of `parquet.ColumnChunk` and `parquet.Page`
+to expose column values of the row group.
+
+This model can be preferable when the underlying storage or in-memory
+representation of the data needs to be optimized further than what can be
+achieved by using an intermediary buffering layer with `parquet.Buffer`.
+
+See [parquet.RowGroup](https://pkg.go.dev/github.com/segmentio/parquet-go#RowGroup)
+for the full interface documentation.
+
 ## Maintenance
 
 The project is hosted and maintained by Twilio; we welcome external contributors
diff --git a/bloom.go b/bloom.go
index 17754db..b123e4f 100644
--- a/bloom.go
+++ b/bloom.go
@@ -22,15 +22,12 @@ type BloomFilter interface {
 
 	// 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)
+	Check(value Value) (bool, error)
 }
 
 type bloomFilter struct {
@@ -39,8 +36,23 @@ type bloomFilter struct {
 	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 (f *bloomFilter) Check(v Value) (bool, error) {
+	return f.check(&f.SectionReader, f.Size(), v.hash(f.hash))
+}
+
+func (v Value) hash(h bloom.Hash) uint64 {
+	switch v.Kind() {
+	case Boolean:
+		return h.Sum64Uint8(uint8(v.u64))
+	case Int32, Float:
+		return h.Sum64Uint32(uint32(v.u64))
+	case Int64, Double:
+		return h.Sum64Uint64(v.u64)
+	case Int96:
+		return h.Sum64(v.Bytes())
+	default:
+		return h.Sum64(v.ByteArray())
+	}
 }
 
 func newBloomFilter(file io.ReaderAt, offset int64, header *format.BloomFilterHeader) *bloomFilter {
@@ -124,10 +136,6 @@ func newBloomFilterEncoder(filter bloom.MutableFilter, hash bloom.Hash) *bloomFi
 	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()
 }
diff --git a/bloom/hash.go b/bloom/hash.go
index d8b7b5c..90df4e0 100644
--- a/bloom/hash.go
+++ b/bloom/hash.go
@@ -8,6 +8,14 @@ import "github.com/segmentio/parquet-go/bloom/xxhash"
 type Hash interface {
 	// Returns the 64 bit hash of the value passed as argument.
 	Sum64(value []byte) uint64
+
+	// Compute hashes of individual values of primitive types.
+	Sum64Uint8(value uint8) uint64
+	Sum64Uint16(value uint16) uint64
+	Sum64Uint32(value uint32) uint64
+	Sum64Uint64(value uint64) uint64
+	Sum64Uint128(value [16]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
@@ -24,6 +32,26 @@ func (XXH64) Sum64(b []byte) uint64 {
 	return xxhash.Sum64(b)
 }
 
+func (XXH64) Sum64Uint8(v uint8) uint64 {
+	return xxhash.Sum64Uint8(v)
+}
+
+func (XXH64) Sum64Uint16(v uint16) uint64 {
+	return xxhash.Sum64Uint16(v)
+}
+
+func (XXH64) Sum64Uint32(v uint32) uint64 {
+	return xxhash.Sum64Uint32(v)
+}
+
+func (XXH64) Sum64Uint64(v uint64) uint64 {
+	return xxhash.Sum64Uint64(v)
+}
+
+func (XXH64) Sum64Uint128(v [16]byte) uint64 {
+	return xxhash.Sum64Uint128(v)
+}
+
 func (XXH64) MultiSum64Uint8(h []uint64, v []uint8) int {
 	return xxhash.MultiSum64Uint8(h, v)
 }
diff --git a/column_buffer.go b/column_buffer.go
index 88f7341..6b96eaa 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -455,10 +455,10 @@ func (col *repeatedColumnBuffer) Page() BufferedPage {
 			for i := int64(0); i < numValues; i++ {
 				var err error
 				if buffer, err = col.base.ReadRowAt(buffer[:0], int64(row.offset)+i); err != nil {
-					return newErrorPage(int16(col.Column()), "reordering rows of repeated column: %w", err)
+					return newErrorPage(col.Column(), "reordering rows of repeated column: %w", err)
 				}
 				if err = column.base.WriteRow(buffer); err != nil {
-					return newErrorPage(int16(col.Column()), "reordering rows of repeated column: %w", err)
+					return newErrorPage(col.Column(), "reordering rows of repeated column: %w", err)
 				}
 			}
 		}
diff --git a/column_reader.go b/column_reader.go
new file mode 100644
index 0000000..105d287
--- /dev/null
+++ b/column_reader.go
@@ -0,0 +1,869 @@
+package parquet
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/encoding/plain"
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+// ColumnReader is an interface implemented by types which support reading
+// columns of values. The interface extends ValueReader to work on top of
+// parquet encodings.
+//
+// Implementations of ColumnReader may also provide extensions that the
+// application can detect using type assertions. For example, readers for
+// columns of INT32 values may implement the parquet.Int32Reader interface
+// as a mechanism to provide a type safe and more efficient access to the
+// column values.
+type ColumnReader interface {
+	ValueReader
+
+	// Returns the type of values read.
+	Type() Type
+
+	// Returns the column number of values read.
+	Column() int
+
+	// Resets the reader state to read values from the given decoder.
+	//
+	// Column readers created from parquet types are initialized to an empty
+	// state and will return io.EOF on every read until a decoder is installed
+	// via a call to Reset.
+	Reset(decoder encoding.Decoder)
+}
+
+type columnReader struct {
+	remain             int
+	numValues          int
+	maxRepetitionLevel int8
+	maxDefinitionLevel int8
+	repetitions        levelReader
+	definitions        levelReader
+	values             ColumnReader
+}
+
+func newColumnReader(values ColumnReader, maxRepetitionLevel, maxDefinitionLevel int8, bufferSize int) *columnReader {
+	repetitionBufferSize := 0
+	definitionBufferSize := 0
+
+	switch {
+	case maxRepetitionLevel > 0 && maxDefinitionLevel > 0:
+		repetitionBufferSize = bufferSize / 2
+		definitionBufferSize = bufferSize / 2
+
+	case maxRepetitionLevel > 0:
+		repetitionBufferSize = bufferSize
+
+	case maxDefinitionLevel > 0:
+		definitionBufferSize = bufferSize
+	}
+
+	return &columnReader{
+		maxRepetitionLevel: maxRepetitionLevel,
+		maxDefinitionLevel: maxDefinitionLevel,
+		repetitions:        makeLevelReader(repetitionBufferSize),
+		definitions:        makeLevelReader(definitionBufferSize),
+		values:             values,
+	}
+}
+
+func (r *columnReader) Type() Type { return r.values.Type() }
+
+func (r *columnReader) Column() int { return r.values.Column() }
+
+func (r *columnReader) ReadValues(values []Value) (int, error) {
+	if r.values == nil {
+		return 0, io.EOF
+	}
+	read := 0
+	columnIndex := ^int16(r.Column())
+
+	for r.remain > 0 && len(values) > 0 {
+		var err error
+		var repetitionLevels []int8
+		var definitionLevels []int8
+		var numValues = r.remain
+
+		if len(values) < numValues {
+			numValues = len(values)
+		}
+
+		if r.maxRepetitionLevel > 0 {
+			repetitionLevels, err = r.repetitions.peekLevels()
+			if err != nil {
+				return read, fmt.Errorf("decoding repetition level from data page of column %d: %w", r.Column(), err)
+			}
+			if len(repetitionLevels) < numValues {
+				numValues = len(repetitionLevels)
+			}
+		}
+
+		if r.maxDefinitionLevel > 0 {
+			definitionLevels, err = r.definitions.peekLevels()
+			if err != nil {
+				return read, fmt.Errorf("decoding definition level from data page of column %d: %w", r.Column(), err)
+			}
+			if len(definitionLevels) < numValues {
+				numValues = len(definitionLevels)
+			}
+		}
+
+		if len(repetitionLevels) > 0 {
+			repetitionLevels = repetitionLevels[:numValues]
+		}
+		if len(definitionLevels) > 0 {
+			definitionLevels = definitionLevels[:numValues]
+		}
+		numNulls := countLevelsNotEqual(definitionLevels, r.maxDefinitionLevel)
+		wantRead := numValues - numNulls
+		n, err := r.values.ReadValues(values[:wantRead])
+		if n < wantRead && err != nil {
+			if err == io.EOF {
+				// EOF should not happen at this stage since we successfully
+				// decoded levels.
+				err = fmt.Errorf("after reading %d/%d values: %w", r.numValues-r.remain, r.numValues, io.ErrUnexpectedEOF)
+			}
+			return read, fmt.Errorf("decoding values from data page of column %d: %w", r.Column(), err)
+		}
+
+		for i, j := n-1, len(definitionLevels)-1; j >= 0; j-- {
+			if definitionLevels[j] != r.maxDefinitionLevel {
+				values[j] = Value{columnIndex: columnIndex}
+			} else {
+				values[j] = values[i]
+				i--
+			}
+		}
+
+		for i, lvl := range repetitionLevels {
+			values[i].repetitionLevel = lvl
+		}
+
+		for i, lvl := range definitionLevels {
+			values[i].definitionLevel = lvl
+		}
+
+		values = values[numValues:]
+		r.repetitions.discardLevels(len(repetitionLevels))
+		r.definitions.discardLevels(len(definitionLevels))
+		r.remain -= numValues
+		read += numValues
+	}
+
+	if r.remain == 0 {
+		return read, io.EOF
+	}
+
+	return read, nil
+}
+
+func (r *columnReader) reset(numValues int, repetitions, definitions, values encoding.Decoder) {
+	if repetitions != nil {
+		repetitions.SetBitWidth(bits.Len8(r.maxRepetitionLevel))
+	}
+	if definitions != nil {
+		definitions.SetBitWidth(bits.Len8(r.maxDefinitionLevel))
+	}
+	r.remain = numValues
+	r.numValues = numValues
+	r.repetitions.reset(repetitions)
+	r.definitions.reset(definitions)
+	r.values.Reset(values)
+}
+
+func (r *columnReader) Reset(encoding.Decoder) {
+	panic("BUG: parquet.columnReader.Reset must not be called")
+}
+
+type levelReader struct {
+	decoder encoding.Decoder
+	levels  []int8
+	offset  int
+	count   int
+}
+
+func makeLevelReader(bufferSize int) levelReader {
+	return levelReader{
+		levels: make([]int8, 0, bufferSize),
+	}
+}
+
+func (r *levelReader) readLevel() (int8, error) {
+	for {
+		if r.offset < len(r.levels) {
+			lvl := r.levels[r.offset]
+			r.offset++
+			return lvl, nil
+		}
+		if err := r.decodeLevels(); err != nil {
+			return -1, err
+		}
+	}
+}
+
+func (r *levelReader) peekLevels() ([]int8, error) {
+	if r.offset == len(r.levels) {
+		if err := r.decodeLevels(); err != nil {
+			return nil, err
+		}
+	}
+	return r.levels[r.offset:], nil
+}
+
+func (r *levelReader) discardLevels(n int) {
+	remain := len(r.levels) - r.offset
+	switch {
+	case n > remain:
+		panic("BUG: cannot discard more levels than buffered")
+	case n == remain:
+		r.levels = r.levels[:0]
+		r.offset = 0
+	default:
+		r.offset += n
+	}
+}
+
+func (r *levelReader) decodeLevels() error {
+	n, err := r.decoder.DecodeInt8(r.levels[:cap(r.levels)])
+	if n == 0 {
+		return err
+	}
+	r.levels = r.levels[:n]
+	r.offset = 0
+	r.count += n
+	return nil
+}
+
+func (r *levelReader) reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.levels = r.levels[:0]
+	r.offset = 0
+	r.count = 0
+}
+
+// The types below are implementations of the ColumnReader interface for each
+// primitive type supported by parquet.
+//
+// The readers use an in-memory intermediary buffer to support decoding arrays
+// of values from the underlying decoder, which are then boxed into the []Value
+// buffer passed to ReadValues. When the program converts type checks the
+// readers for more specific interfaces (e.g. parquet.Int32Reader), the values
+// can be decoded directly from the underlying decoder. There is no need for
+// the intermediary buffers so they are lazily allocated only if the ReadValues
+// methods are called.
+
+type booleanColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []bool
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newBooleanColumnReader(typ Type, columnIndex int16, bufferSize int) *booleanColumnReader {
+	return &booleanColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *booleanColumnReader) Type() Type { return r.typ }
+
+func (r *booleanColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *booleanColumnReader) ReadBooleans(values []bool) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		return n, io.EOF
+	}
+	d, err := r.decoder.DecodeBoolean(values)
+	return n + d, err
+}
+
+func (r *booleanColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]bool, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueBoolean(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeBoolean(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *booleanColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type int32ColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []int32
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newInt32ColumnReader(typ Type, columnIndex int16, bufferSize int) *int32ColumnReader {
+	return &int32ColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *int32ColumnReader) Type() Type { return r.typ }
+
+func (r *int32ColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *int32ColumnReader) ReadInt32s(values []int32) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeInt32(values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *int32ColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]int32, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueInt32(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeInt32(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *int32ColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type int64ColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []int64
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newInt64ColumnReader(typ Type, columnIndex int16, bufferSize int) *int64ColumnReader {
+	return &int64ColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *int64ColumnReader) Type() Type { return r.typ }
+
+func (r *int64ColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *int64ColumnReader) ReadInt64s(values []int64) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeInt64(values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *int64ColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]int64, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueInt64(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeInt64(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *int64ColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type int96ColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []deprecated.Int96
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newInt96ColumnReader(typ Type, columnIndex int16, bufferSize int) *int96ColumnReader {
+	return &int96ColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *int96ColumnReader) Type() Type { return r.typ }
+
+func (r *int96ColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *int96ColumnReader) ReadInt96s(values []deprecated.Int96) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeInt96(values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *int96ColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]deprecated.Int96, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueInt96(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeInt96(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *int96ColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type floatColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []float32
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newFloatColumnReader(typ Type, columnIndex int16, bufferSize int) *floatColumnReader {
+	return &floatColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *floatColumnReader) Type() Type { return r.typ }
+
+func (r *floatColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *floatColumnReader) ReadFloats(values []float32) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeFloat(values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *floatColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]float32, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueFloat(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeFloat(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *floatColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type doubleColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []float64
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newDoubleColumnReader(typ Type, columnIndex int16, bufferSize int) *doubleColumnReader {
+	return &doubleColumnReader{
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *doubleColumnReader) Type() Type { return r.typ }
+
+func (r *doubleColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *doubleColumnReader) ReadDoubles(values []float64) (n int, err error) {
+	if r.offset < len(r.buffer) {
+		n = copy(values, r.buffer[r.offset:])
+		r.offset += n
+		values = values[n:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeDouble(values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *doubleColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]float64, 0, atLeastOne(r.bufferSize))
+	}
+
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValueDouble(r.buffer[r.offset])
+			values[n].columnIndex = r.columnIndex
+			r.offset++
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeDouble(buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *doubleColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+type byteArrayColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      encoding.ByteArrayList
+	offset      int
+	columnIndex int16
+}
+
+func newByteArrayColumnReader(typ Type, columnIndex int16, bufferSize int) *byteArrayColumnReader {
+	return &byteArrayColumnReader{
+		typ:         typ,
+		buffer:      encoding.MakeByteArrayList(atLeastOne(bufferSize / 16)),
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *byteArrayColumnReader) Type() Type { return r.typ }
+
+func (r *byteArrayColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *byteArrayColumnReader) readByteArrays(do func([]byte) bool) (n int, err error) {
+	for {
+		for r.offset < r.buffer.Len() {
+			if !do(r.buffer.Index(r.offset)) {
+				return n, nil
+			}
+			r.offset++
+			n++
+		}
+
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		r.buffer.Reset()
+		r.offset = 0
+
+		d, err := r.decoder.DecodeByteArray(&r.buffer)
+		if d == 0 {
+			return n, err
+		}
+	}
+}
+
+func (r *byteArrayColumnReader) ReadByteArrays(values []byte) (int, error) {
+	i := 0
+	n, err := r.readByteArrays(func(b []byte) bool {
+		k := plain.ByteArrayLengthSize + len(b)
+		if k > (len(values) - i) {
+			return false
+		}
+		plain.PutByteArrayLength(values[i:], len(b))
+		copy(values[i+plain.ByteArrayLengthSize:], b)
+		i += k
+		return true
+	})
+	if i == 0 && err == nil {
+		err = io.ErrShortBuffer
+	}
+	return n, err
+}
+
+func (r *byteArrayColumnReader) ReadValues(values []Value) (int, error) {
+	i := 0
+	return r.readByteArrays(func(b []byte) (ok bool) {
+		if ok = i < len(values); ok {
+			values[i] = makeValueBytes(ByteArray, copyBytes(b))
+			values[i].columnIndex = r.columnIndex
+			i++
+		}
+		return ok
+	})
+}
+
+func (r *byteArrayColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer.Reset()
+	r.offset = 0
+}
+
+type fixedLenByteArrayColumnReader struct {
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []byte
+	offset      int
+	size        int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newFixedLenByteArrayColumnReader(typ Type, columnIndex int16, bufferSize int) *fixedLenByteArrayColumnReader {
+	return &fixedLenByteArrayColumnReader{
+		typ:         typ,
+		size:        typ.Length(),
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *fixedLenByteArrayColumnReader) Type() Type { return r.typ }
+
+func (r *fixedLenByteArrayColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *fixedLenByteArrayColumnReader) ReadFixedLenByteArrays(values []byte) (n int, err error) {
+	if (len(values) % r.size) != 0 {
+		return 0, fmt.Errorf("cannot read FIXED_LEN_BYTE_ARRAY values of size %d into buffer of size %d", r.size, len(values))
+	}
+	if r.offset < len(r.buffer) {
+		i := copy(values, r.buffer[r.offset:])
+		n += i / r.size
+		r.offset += n
+		values = values[i:]
+	}
+	if r.decoder == nil {
+		err = io.EOF
+	} else {
+		var d int
+		d, err = r.decoder.DecodeFixedLenByteArray(r.size, values)
+		n += d
+	}
+	return n, err
+}
+
+func (r *fixedLenByteArrayColumnReader) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]byte, 0, atLeast((r.bufferSize/r.size)*r.size, r.size))
+	}
+
+	for {
+		for (r.offset+r.size) <= len(r.buffer) && n < len(values) {
+			values[n] = makeValueBytes(FixedLenByteArray, copyBytes(r.buffer[r.offset:r.offset+r.size]))
+			values[n].columnIndex = r.columnIndex
+			r.offset += r.size
+			n++
+		}
+
+		if n == len(values) {
+			return n, nil
+		}
+		if r.decoder == nil {
+			return n, io.EOF
+		}
+
+		buffer := r.buffer[:cap(r.buffer)]
+		d, err := r.decoder.DecodeFixedLenByteArray(r.size, buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d*r.size]
+		r.offset = 0
+	}
+}
+
+func (r *fixedLenByteArrayColumnReader) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
+
+var (
+	_ BooleanReader           = (*booleanColumnReader)(nil)
+	_ Int32Reader             = (*int32ColumnReader)(nil)
+	_ Int64Reader             = (*int64ColumnReader)(nil)
+	_ Int96Reader             = (*int96ColumnReader)(nil)
+	_ FloatReader             = (*floatColumnReader)(nil)
+	_ DoubleReader            = (*doubleColumnReader)(nil)
+	_ ByteArrayReader         = (*byteArrayColumnReader)(nil)
+	_ FixedLenByteArrayReader = (*fixedLenByteArrayColumnReader)(nil)
+)
diff --git a/concat.go b/concat.go
index a35921e..33ae2df 100644
--- a/concat.go
+++ b/concat.go
@@ -145,10 +145,10 @@ func (f concatenatedBloomFilter) Size() int64 {
 	return size
 }
 
-func (f concatenatedBloomFilter) Check(key []byte) (bool, error) {
+func (f concatenatedBloomFilter) Check(v Value) (bool, error) {
 	for _, c := range f.chunks {
 		if b := c.BloomFilter(); b != nil {
-			if ok, err := b.Check(key); ok || err != nil {
+			if ok, err := b.Check(v); ok || err != nil {
 				return ok, err
 			}
 		}
diff --git a/convert.go b/convert.go
index 9441490..7dae066 100644
--- a/convert.go
+++ b/convert.go
@@ -62,7 +62,7 @@ func (c *conversion) Schema() *Schema { return c.schema }
 // null or zero values.
 //
 // The returned function is intended to be used to append the converted source
-// row to the destinatination buffer.
+// row to the destination buffer.
 func Convert(to, from Node) (conv Conversion, err error) {
 	defer func() {
 		switch e := recover().(type) {
@@ -497,7 +497,7 @@ 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 }
+func (missingBloomFilter) Check(Value) (bool, error)         { return false, nil }
 
 type missingPage struct{ *missingColumnChunk }
 
@@ -510,7 +510,7 @@ func (p missingPage) Bounds() (min, max Value) { return }
 func (p missingPage) Size() int64              { return 0 }
 func (p missingPage) Values() ValueReader      { return &missingValues{page: p} }
 func (p missingPage) Buffer() BufferedPage {
-	return newErrorPage(p.column, "cannot buffer missing page")
+	return newErrorPage(p.Column(), "cannot buffer missing page")
 }
 
 type missingValues struct {
diff --git a/dictionary.go b/dictionary.go
index dfce3b3..b9b0573 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -687,8 +687,8 @@ func (t *indexedType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer
 	return newIndexedColumnBuffer(t.dict, t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *indexedType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newIndexedValueDecoder(t.dict, bufferSize)
+func (t *indexedType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newIndexedColumnReader(t.dict, t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type indexedPage struct {
@@ -762,14 +762,18 @@ type indexedPageReader struct {
 }
 
 func (r *indexedPageReader) ReadValues(values []Value) (n int, err error) {
+	var v Value
 	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = r.page.dict.Index(int(r.page.values[r.offset]))
+		v = r.page.dict.Index(int(r.page.values[r.offset]))
+		v.columnIndex = r.page.columnIndex
+		values[n] = v
 		r.offset++
 		n++
 	}
 	if r.offset == len(r.page.values) {
 		err = io.EOF
 	}
+
 	return n, err
 }
 
@@ -861,59 +865,40 @@ func (col *indexedColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	}
 }
 
-type indexedColumnIndex struct{ col *indexedColumnBuffer }
-
-func (index indexedColumnIndex) NumPages() int       { return 1 }
-func (index indexedColumnIndex) NullCount(int) int64 { return 0 }
-func (index indexedColumnIndex) NullPage(int) bool   { return false }
-func (index indexedColumnIndex) MinValue(int) []byte {
-	min, _ := index.col.Bounds()
-	return min.Bytes()
-}
-func (index indexedColumnIndex) MaxValue(int) []byte {
-	_, max := index.col.Bounds()
-	return max.Bytes()
-}
-func (index indexedColumnIndex) IsAscending() bool {
-	return index.col.typ.Compare(index.col.Bounds()) < 0
-}
-func (index indexedColumnIndex) IsDescending() bool {
-	return index.col.typ.Compare(index.col.Bounds()) > 0
-}
-
-type indexedOffsetIndex struct{ col *indexedColumnBuffer }
-
-func (index indexedOffsetIndex) NumPages() int                { return 1 }
-func (index indexedOffsetIndex) Offset(int) int64             { return 0 }
-func (index indexedOffsetIndex) CompressedPageSize(int) int64 { return index.col.Size() }
-func (index indexedOffsetIndex) FirstRowIndex(int) int64      { return 0 }
-
-type indexedValueDecoder struct {
-	decoder encoding.Decoder
-	dict    Dictionary
-	values  []int32
-	offset  uint
+type indexedColumnReader struct {
+	dict        Dictionary
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []int32
+	offset      int
+	columnIndex int16
 }
 
-func newIndexedValueDecoder(dict Dictionary, bufferSize int) *indexedValueDecoder {
-	return &indexedValueDecoder{
-		dict:   dict,
-		values: make([]int32, 0, atLeastOne(bufferSize/4)),
+func newIndexedColumnReader(dict Dictionary, typ Type, columnIndex int16, bufferSize int) *indexedColumnReader {
+	return &indexedColumnReader{
+		dict:        dict,
+		typ:         typ,
+		buffer:      make([]int32, 0, atLeastOne(bufferSize)),
+		columnIndex: ^columnIndex,
 	}
 }
 
-func (r *indexedValueDecoder) ReadValues(values []Value) (int, error) {
+func (r *indexedColumnReader) Type() Type { return r.typ }
+
+func (r *indexedColumnReader) Column() int { return int(^r.columnIndex) }
+
+func (r *indexedColumnReader) ReadValues(values []Value) (int, error) {
 	i := 0
 	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			count := uint(len(r.values)) - r.offset
-			limit := uint(len(values) - i)
+		for r.offset < len(r.buffer) && i < len(values) {
+			count := len(r.buffer) - r.offset
+			limit := len(values) - i
 
 			if count > limit {
 				count = limit
 			}
 
-			indexes := r.values[r.offset : r.offset+count]
+			indexes := r.buffer[r.offset : r.offset+count]
 			dictLen := r.dict.Len()
 			for _, index := range indexes {
 				if index < 0 || int(index) >= dictLen {
@@ -923,29 +908,59 @@ func (r *indexedValueDecoder) ReadValues(values []Value) (int, error) {
 
 			r.dict.Lookup(indexes, values[i:])
 			r.offset += count
+
+			j := i
 			i += int(count)
+			for j < i {
+				values[j].columnIndex = r.columnIndex
+				j++
+			}
 		}
 
 		if i == len(values) {
 			return i, nil
 		}
 
-		n, err := r.decoder.DecodeInt32(r.values[:cap(r.values)])
+		buffer := r.buffer[:cap(r.buffer)]
+		n, err := r.decoder.DecodeInt32(buffer)
 		if n == 0 {
 			return i, err
 		}
 
-		r.values = r.values[:n]
+		r.buffer = buffer[:n]
 		r.offset = 0
 	}
 }
 
-func (r *indexedValueDecoder) Reset(decoder encoding.Decoder) {
+func (r *indexedColumnReader) Reset(decoder encoding.Decoder) {
 	r.decoder = decoder
-	r.values = r.values[:0]
+	r.buffer = r.buffer[:0]
 	r.offset = 0
 }
 
-func (r *indexedValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
+type indexedColumnIndex struct{ col *indexedColumnBuffer }
+
+func (index indexedColumnIndex) NumPages() int       { return 1 }
+func (index indexedColumnIndex) NullCount(int) int64 { return 0 }
+func (index indexedColumnIndex) NullPage(int) bool   { return false }
+func (index indexedColumnIndex) MinValue(int) []byte {
+	min, _ := index.col.Bounds()
+	return min.Bytes()
 }
+func (index indexedColumnIndex) MaxValue(int) []byte {
+	_, max := index.col.Bounds()
+	return max.Bytes()
+}
+func (index indexedColumnIndex) IsAscending() bool {
+	return index.col.typ.Compare(index.col.Bounds()) < 0
+}
+func (index indexedColumnIndex) IsDescending() bool {
+	return index.col.typ.Compare(index.col.Bounds()) > 0
+}
+
+type indexedOffsetIndex struct{ col *indexedColumnBuffer }
+
+func (index indexedOffsetIndex) NumPages() int                { return 1 }
+func (index indexedOffsetIndex) Offset(int) int64             { return 0 }
+func (index indexedOffsetIndex) CompressedPageSize(int) int64 { return index.col.Size() }
+func (index indexedOffsetIndex) FirstRowIndex(int) int64      { return 0 }
diff --git a/file.go b/file.go
index c1443ff..ca14841 100644
--- a/file.go
+++ b/file.go
@@ -814,7 +814,7 @@ 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: int16(p.Column())}
+		return &errorPage{err: err, columnIndex: p.Column()}
 	}
 	return bufferedPage.Page()
 }
@@ -839,7 +839,7 @@ func (p *filePage) PageSize() int64 { return int64(p.header.CompressedPageSize)
 func (p *filePage) CRC() uint32 { return uint32(p.header.CRC) }
 
 type filePageValueReaderState struct {
-	reader *dataPageReader
+	reader ColumnReader
 
 	v1 struct {
 		repetitions dataPageLevelV1
@@ -912,30 +912,40 @@ func (s *filePageValueReaderState) init(columnType Type, column *Column, codec f
 		return fmt.Errorf("cannot read values from page of type %s", h.PageType())
 	}
 
-	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
 
+	maxRepetitionLevel := column.maxRepetitionLevel
+	maxDefinitionLevel := column.maxDefinitionLevel
+	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 {
-		s.reader = newDataPageReader(
-			columnType,
-			column.maxRepetitionLevel,
-			column.maxDefinitionLevel,
-			column.index,
-			defaultReadBufferSize,
-		)
+		bufferSize := defaultReadBufferSize
+		if hasLevels {
+			bufferSize /= 2
+		}
+		s.reader = columnType.NewColumnReader(int(column.index), bufferSize)
+		if hasLevels {
+			s.reader = newColumnReader(s.reader, maxRepetitionLevel, maxDefinitionLevel, bufferSize)
+		}
+	}
+
+	switch r := s.reader.(type) {
+	case *columnReader:
+		r.reset(int(header.NumValues()), s.repetitions.decoder, s.definitions.decoder, s.page.decoder)
+	default:
+		r.Reset(s.page.decoder)
 	}
 
-	// TODO: revisit the data page reader APIs
-	s.reader.Reset(int(header.NumValues()), s.repetitions.decoder, s.definitions.decoder, s.page.decoder)
 	return nil
 }
 
diff --git a/internal/bits/max_amd64.s b/internal/bits/max_amd64.s
index 3a152c4..f5d6b50 100644
--- a/internal/bits/max_amd64.s
+++ b/internal/bits/max_amd64.s
@@ -459,7 +459,7 @@ TEXT ·maxBE128(SB), NOSPLIT, $-48
     JB loop
 
     CMPB ·hasAVX512MinMaxBE128(SB), $0
-    JB loop
+    JE loop
 
     // Z19 holds a vector of the count by which we increment the vectors of
     // swap at each loop iteration.
diff --git a/internal/bits/min_amd64.s b/internal/bits/min_amd64.s
index 07ec94c..5c4aa12 100644
--- a/internal/bits/min_amd64.s
+++ b/internal/bits/min_amd64.s
@@ -460,7 +460,7 @@ TEXT ·minBE128(SB), NOSPLIT, $-48
     JB loop
 
     CMPB ·hasAVX512MinMaxBE128(SB), $0
-    JB loop
+    JE loop
 
     // Z19 holds a vector of the count by which we increment the vectors of
     // swap at each loop iteration.
diff --git a/page.go b/page.go
index fcbca9c..6015a92 100644
--- a/page.go
+++ b/page.go
@@ -45,6 +45,12 @@ type Page interface {
 	Size() int64
 
 	// Returns a reader exposing the values contained in the page.
+	//
+	// Depending on the underlying implementation, the returned reader may
+	// support reading an array of typed Go values by implementing interfaces
+	// like parquet.Int32Reader. Applications should use type assertions on
+	// the returned reader to determine whether those optimizations are
+	// available.
 	Values() ValueReader
 
 	// Buffer returns the page as a BufferedPage, which may be the page itself
@@ -199,17 +205,17 @@ func forEachPageSlice(page BufferedPage, wantSize int64, do func(BufferedPage) e
 
 type errorPage struct {
 	err         error
-	columnIndex int16
+	columnIndex int
 }
 
-func newErrorPage(columnIndex int16, msg string, args ...interface{}) *errorPage {
+func newErrorPage(columnIndex int, msg string, args ...interface{}) *errorPage {
 	return &errorPage{
 		err:         fmt.Errorf(msg, args...),
 		columnIndex: columnIndex,
 	}
 }
 
-func (page *errorPage) Column() int                    { return int(page.columnIndex) }
+func (page *errorPage) Column() int                    { return page.columnIndex }
 func (page *errorPage) Dictionary() Dictionary         { return nil }
 func (page *errorPage) NumRows() int64                 { return 0 }
 func (page *errorPage) NumValues() int64               { return 0 }
@@ -368,7 +374,8 @@ func (r *optionalPageReader) ReadValues(values []Value) (n int, err error) {
 				r.offset++
 				n++
 			}
-			if err != nil {
+			// Do not return on an io.EOF here as we may still have null values to read.
+			if err != nil && err != io.EOF {
 				return n, err
 			}
 		}
diff --git a/page_reader.go b/page_reader.go
deleted file mode 100644
index 3782508..0000000
--- a/page_reader.go
+++ /dev/null
@@ -1,222 +0,0 @@
-package parquet
-
-import (
-	"fmt"
-	"io"
-
-	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/internal/bits"
-)
-
-// dataPageReader reads values from a data page.
-//
-// dataPageReader implements the ValueReader interface; when they exist,
-// the reader decodes repetition and definition levels in order to assign
-// levels to values returned to the application, which includes producing
-// null values when needed.
-type dataPageReader struct {
-	remain             int
-	numValues          int
-	maxRepetitionLevel int8
-	maxDefinitionLevel int8
-	columnIndex        int16
-	repetitions        levelReader
-	definitions        levelReader
-	values             ValueDecoder
-}
-
-func newDataPageReader(typ Type, maxRepetitionLevel, maxDefinitionLevel int8, columnIndex int16, bufferSize int) *dataPageReader {
-	bufferSize /= 2
-	repetitionBufferSize := 0
-	definitionBufferSize := 0
-
-	switch {
-	case maxRepetitionLevel > 0 && maxDefinitionLevel > 0:
-		repetitionBufferSize = bufferSize / 2
-		definitionBufferSize = bufferSize / 2
-
-	case maxRepetitionLevel > 0:
-		repetitionBufferSize = bufferSize
-
-	case maxDefinitionLevel > 0:
-		definitionBufferSize = bufferSize
-	}
-
-	return &dataPageReader{
-		maxRepetitionLevel: maxRepetitionLevel,
-		maxDefinitionLevel: maxDefinitionLevel,
-		columnIndex:        ^columnIndex,
-		repetitions:        makeLevelReader(repetitionBufferSize),
-		definitions:        makeLevelReader(definitionBufferSize),
-		values:             typ.NewValueDecoder(bufferSize),
-	}
-}
-
-func (r *dataPageReader) Reset(numValues int, repetitions, definitions, values encoding.Decoder) {
-	if repetitions != nil {
-		repetitions.SetBitWidth(bits.Len8(r.maxRepetitionLevel))
-	}
-	if definitions != nil {
-		definitions.SetBitWidth(bits.Len8(r.maxDefinitionLevel))
-	}
-	r.remain = numValues
-	r.numValues = numValues
-	r.repetitions.reset(repetitions)
-	r.definitions.reset(definitions)
-	r.values.Reset(values)
-}
-
-func (r *dataPageReader) ReadValues(values []Value) (int, error) {
-	if r.values == nil {
-		return 0, io.EOF
-	}
-	read := 0
-
-	for r.remain > 0 && len(values) > 0 {
-		var err error
-		var repetitionLevels []int8
-		var definitionLevels []int8
-		var numValues = r.remain
-
-		if len(values) < numValues {
-			numValues = len(values)
-		}
-
-		if r.maxRepetitionLevel > 0 {
-			repetitionLevels, err = r.repetitions.peekLevels()
-			if err != nil {
-				return read, fmt.Errorf("decoding repetition level from data page of column %d: %w", ^r.columnIndex, err)
-			}
-			if len(repetitionLevels) < numValues {
-				numValues = len(repetitionLevels)
-			}
-		}
-
-		if r.maxDefinitionLevel > 0 {
-			definitionLevels, err = r.definitions.peekLevels()
-			if err != nil {
-				return read, fmt.Errorf("decoding definition level from data page of column %d: %w", ^r.columnIndex, err)
-			}
-			if len(definitionLevels) < numValues {
-				numValues = len(definitionLevels)
-			}
-		}
-
-		if len(repetitionLevels) > 0 {
-			repetitionLevels = repetitionLevels[:numValues]
-		}
-		if len(definitionLevels) > 0 {
-			definitionLevels = definitionLevels[:numValues]
-		}
-		numNulls := countLevelsNotEqual(definitionLevels, r.maxDefinitionLevel)
-		wantRead := numValues - numNulls
-		n, err := r.values.ReadValues(values[:wantRead])
-		if n < wantRead && err != nil {
-			if err == io.EOF {
-				// EOF should not happen at this stage since we successfully
-				// decoded levels.
-				err = fmt.Errorf("after reading %d/%d values: %w", r.numValues-r.remain, r.numValues, io.ErrUnexpectedEOF)
-			}
-			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-- {
-			if definitionLevels[j] != r.maxDefinitionLevel {
-				values[j] = Value{}
-			} else {
-				values[j] = values[i]
-				i--
-			}
-		}
-
-		for i, lvl := range repetitionLevels {
-			values[i].repetitionLevel = lvl
-		}
-
-		for i, lvl := range definitionLevels {
-			values[i].definitionLevel = lvl
-		}
-
-		for i := range values[:numValues] {
-			values[i].columnIndex = r.columnIndex
-		}
-
-		values = values[numValues:]
-		r.repetitions.discardLevels(len(repetitionLevels))
-		r.definitions.discardLevels(len(definitionLevels))
-		r.remain -= numValues
-		read += numValues
-	}
-
-	if r.remain == 0 {
-		return read, io.EOF
-	}
-
-	return read, nil
-}
-
-type levelReader struct {
-	decoder encoding.Decoder
-	levels  []int8
-	offset  uint
-	count   uint
-}
-
-func makeLevelReader(bufferSize int) levelReader {
-	return levelReader{
-		levels: make([]int8, 0, bufferSize),
-	}
-}
-
-func (r *levelReader) readLevel() (int8, error) {
-	for {
-		if r.offset < uint(len(r.levels)) {
-			lvl := r.levels[r.offset]
-			r.offset++
-			return lvl, nil
-		}
-		if err := r.decodeLevels(); err != nil {
-			return -1, err
-		}
-	}
-}
-
-func (r *levelReader) peekLevels() ([]int8, error) {
-	if r.offset == uint(len(r.levels)) {
-		if err := r.decodeLevels(); err != nil {
-			return nil, err
-		}
-	}
-	return r.levels[r.offset:], nil
-}
-
-func (r *levelReader) discardLevels(n int) {
-	remain := uint(len(r.levels)) - r.offset
-	switch {
-	case uint(n) > remain:
-		panic("BUG: cannot discard more levels than buffered")
-	case uint(n) == remain:
-		r.levels = r.levels[:0]
-		r.offset = 0
-	default:
-		r.offset += uint(n)
-	}
-}
-
-func (r *levelReader) decodeLevels() error {
-	n, err := r.decoder.DecodeInt8(r.levels[:cap(r.levels)])
-	if n == 0 {
-		return err
-	}
-	r.levels = r.levels[:n]
-	r.offset = 0
-	r.count += uint(n)
-	return nil
-}
-
-func (r *levelReader) reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.levels = r.levels[:0]
-	r.offset = 0
-	r.count = 0
-}
diff --git a/parquet.go b/parquet.go
index 5bc722a..e7d9e74 100644
--- a/parquet.go
+++ b/parquet.go
@@ -6,3 +6,14 @@
 //
 // Or see the Parquet documentation: https://parquet.apache.org/documentation/latest/
 package parquet
+
+func atLeastOne(size int) int {
+	return atLeast(size, 1)
+}
+
+func atLeast(size, least int) int {
+	if size < least {
+		return least
+	}
+	return size
+}
diff --git a/row_group.go b/row_group.go
index 0024a3b..18b95d3 100644
--- a/row_group.go
+++ b/row_group.go
@@ -19,6 +19,12 @@ type RowGroup interface {
 	NumColumns() int
 
 	// Returns the leaf column at the given index in the group.
+	//
+	// If the underlying implementation is not read-only, the returned
+	// parquet.ColumnChunk may implement other interfaces: for example,
+	// parquet.ColumnBuffer if the chunk is backed by an in-memory buffer,
+	// or typed writer interfaces like parquet.Int32Writer depending on the
+	// underlying type of values that can be written to the chunk.
 	Column(int) ColumnChunk
 
 	// Returns the schema of rows in the group.
@@ -31,6 +37,14 @@ type RowGroup interface {
 	SortingColumns() []SortingColumn
 
 	// Returns a reader exposing the rows of the row group.
+	//
+	// As an optimization, the returned parquet.Rows object may implement
+	// parquet.RowWriterTo, and test the RowWriter it receives for an
+	// implementation of the parquet.RowGroupWriter interface.
+	//
+	// This optimization mechanism is leveraged by the parquet.CopyRows function
+	// to skip the generic row-by-row copy algorithm and delegate the copy logic
+	// to the parquet.Rows object.
 	Rows() Rows
 }
 
@@ -178,7 +192,7 @@ func MergeRowGroups(rowGroups []RowGroup, options ...RowGroupOption) (RowGroup,
 	if len(m.sorting) == 0 {
 		// When the row group has no ordering, use a simpler version of the
 		// merger which simply concatenates rows from each of the row groups.
-		// This is preferrable because it makes the output deterministic, the
+		// This is preferable because it makes the output deterministic, the
 		// heap merge may otherwise reorder rows across groups.
 		return &m.concatenatedRowGroup, nil
 	}
@@ -222,6 +236,10 @@ func (r *rowGroup) SortingColumns() []SortingColumn { return r.sorting }
 func (r *rowGroup) Schema() *Schema                 { return r.schema }
 func (r *rowGroup) Rows() Rows                      { return &rowGroupRowReader{rowGroup: r} }
 
+func NewRowGroupRowReader(rowGroup RowGroup) Rows {
+	return &rowGroupRowReader{rowGroup: rowGroup}
+}
+
 type rowGroupRowReader struct {
 	rowGroup RowGroup
 	schema   *Schema
@@ -438,7 +456,7 @@ 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 }
+func (emptyBloomFilter) Check(Value) (bool, error)         { return false, nil }
 
 type emptyRowReader struct{ schema *Schema }
 
diff --git a/type.go b/type.go
index 8fb60f2..b95a53d 100644
--- a/type.go
+++ b/type.go
@@ -128,10 +128,21 @@ type Type interface {
 	// The method panics if it is called on a group type.
 	NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer
 
-	// Creates a decoder for values of this type.
+	// Creates a reader for columns of this type.
 	//
-	// The method panics if it is called on a group type.
-	NewValueDecoder(bufferSize int) ValueDecoder
+	// Column readers are created using the index of the column they are reading
+	// values from (relative to the parent schema). The column index will be set
+	// on values read from the reader.
+	//
+	// The buffer size is given in bytes, because we want to control memory
+	// consumption of the application, which is simpler to achieve with buffer
+	// size expressed in bytes rather than number of elements.
+	//
+	// The returned reader may implement extensions that can be tested via type
+	// assertions. For example, on a INT32 type, the reader could implement the
+	// parquet.Int32Reader interface to allow programs to more efficiently read
+	// columns of INT32 values.
+	NewColumnReader(columnIndex, bufferSize int) ColumnReader
 }
 
 // In the current parquet version supported by this library, only type-defined
@@ -212,8 +223,8 @@ func (t booleanType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newBooleanColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t booleanType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newBooleanValueDecoder(bufferSize)
+func (t booleanType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newBooleanColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type int32Type struct{ primitiveType }
@@ -244,8 +255,8 @@ func (t int32Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t int32Type) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newInt32ValueDecoder(bufferSize)
+func (t int32Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type int64Type struct{ primitiveType }
@@ -276,8 +287,8 @@ func (t int64Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t int64Type) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newInt64ValueDecoder(bufferSize)
+func (t int64Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type int96Type struct{ primitiveType }
@@ -308,8 +319,8 @@ func (t int96Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newInt96ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t int96Type) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newInt96ValueDecoder(bufferSize)
+func (t int96Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt96ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type floatType struct{ primitiveType }
@@ -340,8 +351,8 @@ func (t floatType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newFloatColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t floatType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newFloatValueDecoder(bufferSize)
+func (t floatType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFloatColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type doubleType struct{ primitiveType }
@@ -370,8 +381,8 @@ func (t doubleType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newDoubleColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t doubleType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newDoubleValueDecoder(bufferSize)
+func (t doubleType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newDoubleColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type byteArrayType struct{ primitiveType }
@@ -402,8 +413,8 @@ func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer
 	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t byteArrayType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newByteArrayValueDecoder(bufferSize)
+func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 type fixedLenByteArrayType struct {
@@ -439,8 +450,8 @@ func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) Col
 	return newFixedLenByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *fixedLenByteArrayType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newFixedLenByteArrayValueDecoder(t.length, bufferSize)
+func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 var (
@@ -604,11 +615,11 @@ func (t *intType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	}
 }
 
-func (t *intType) NewValueDecoder(bufferSize int) ValueDecoder {
+func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	if t.BitWidth == 64 {
-		return newInt64ValueDecoder(bufferSize)
+		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 	} else {
-		return newInt32ValueDecoder(bufferSize)
+		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 	}
 }
 
@@ -691,8 +702,8 @@ func (t *stringType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *stringType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newByteArrayValueDecoder(bufferSize)
+func (t *stringType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *stringType) GoType() reflect.Type {
@@ -721,7 +732,7 @@ func (t *uuidType) ColumnOrder() *format.ColumnOrder {
 }
 
 func (t *uuidType) PhyiscalType() *format.Type {
-	return &physicalTypes[ByteArray]
+	return &physicalTypes[FixedLenByteArray]
 }
 
 func (t *uuidType) LogicalType() *format.LogicalType {
@@ -742,8 +753,8 @@ func (t *uuidType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newFixedLenByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *uuidType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newFixedLenByteArrayValueDecoder(16, bufferSize)
+func (t *uuidType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *uuidType) GoType() reflect.Type {
@@ -795,8 +806,8 @@ func (t *enumType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *enumType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newByteArrayValueDecoder(bufferSize)
+func (t *enumType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *enumType) GoType() reflect.Type {
@@ -848,8 +859,8 @@ func (t *jsonType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *jsonType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newByteArrayValueDecoder(bufferSize)
+func (t *jsonType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 // BSON constructs a leaf node of BSON logical type.
@@ -897,8 +908,8 @@ func (t *bsonType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *bsonType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newByteArrayValueDecoder(bufferSize)
+func (t *bsonType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 // Date constructs a leaf node of DATE logical type.
@@ -942,8 +953,8 @@ func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *dateType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newInt32ValueDecoder(bufferSize)
+func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 // TimeUnit represents units of time in the parquet type system.
@@ -1070,11 +1081,11 @@ func (t *timeType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	}
 }
 
-func (t *timeType) NewValueDecoder(bufferSize int) ValueDecoder {
+func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	if t.Unit.Millis != nil {
-		return newInt32ValueDecoder(bufferSize)
+		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 	} else {
-		return newInt64ValueDecoder(bufferSize)
+		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 	}
 }
 
@@ -1126,8 +1137,8 @@ func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffe
 	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
-func (t *timestampType) NewValueDecoder(bufferSize int) ValueDecoder {
-	return newInt64ValueDecoder(bufferSize)
+func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 // List constructs a node of LIST logical type.
@@ -1175,8 +1186,8 @@ func (t *listType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	panic("cannot create row group column from parquet LIST type")
 }
 
-func (t *listType) NewValueDecoder(bufferSize int) ValueDecoder {
-	panic("cannot create page reader from parquet LIST type")
+func (t *listType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	panic("cannot create row group column from parquet LIST type")
 }
 
 // Map constructs a node of MAP logical type.
@@ -1229,8 +1240,8 @@ func (t *mapType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
 	panic("cannot create row group column from parquet MAP type")
 }
 
-func (t *mapType) NewValueDecoder(bufferSize int) ValueDecoder {
-	panic("cannot create page reader from parquet MAP type")
+func (t *mapType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	panic("cannot create row group column from parquet MAP type")
 }
 
 type nullType format.NullType
@@ -1265,8 +1276,8 @@ func (t *nullType) NewColumnBuffer(int, int) ColumnBuffer {
 	panic("cannot create row group column from parquet NULL type")
 }
 
-func (t *nullType) NewValueDecoder(int) ValueDecoder {
-	panic("cannot create page reader from parquet NULL type")
+func (t *nullType) NewColumnReader(int, int) ColumnReader {
+	panic("cannot create row group column from parquet NULL type")
 }
 
 type groupType struct{}
@@ -1293,8 +1304,8 @@ func (t groupType) NewColumnBuffer(int, int) ColumnBuffer {
 	panic("cannot create row group column from parquet group")
 }
 
-func (groupType) NewValueDecoder(int) ValueDecoder {
-	panic("cannot create value decoder from parquet group")
+func (t groupType) NewColumnReader(int, int) ColumnReader {
+	panic("cannot create row group column from parquet group")
 }
 
 func (groupType) Length() int { return 0 }
diff --git a/value.go b/value.go
index ae02e83..807cbb1 100644
--- a/value.go
+++ b/value.go
@@ -326,6 +326,8 @@ func ValueOf(v interface{}) Value {
 		return Value{}
 	case uuid.UUID:
 		return makeValueBytes(FixedLenByteArray, value[:])
+	case deprecated.Int96:
+		return makeValueInt96(value)
 	}
 
 	k := Kind(-1)
diff --git a/value_decoder.go b/value_decoder.go
deleted file mode 100644
index 9ac380e..0000000
--- a/value_decoder.go
+++ /dev/null
@@ -1,413 +0,0 @@
-package parquet
-
-import (
-	"io"
-
-	"github.com/segmentio/parquet-go/deprecated"
-	"github.com/segmentio/parquet-go/encoding"
-)
-
-// ValueDecoder is an extension of the ValueReader interface for types that
-// support decoding values from an underlying decoder.
-type ValueDecoder interface {
-	ValueReader
-	Reset(encoding.Decoder)
-	Decoder() encoding.Decoder
-}
-
-type booleanValueDecoder struct {
-	decoder encoding.Decoder
-	values  []bool
-	offset  uint
-}
-
-func newBooleanValueDecoder(bufferSize int) *booleanValueDecoder {
-	return &booleanValueDecoder{
-		values: make([]bool, 0, atLeastOne(bufferSize)),
-	}
-}
-
-func (r *booleanValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueBoolean(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeBoolean(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *booleanValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *booleanValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type int32ValueDecoder struct {
-	decoder encoding.Decoder
-	values  []int32
-	offset  uint
-}
-
-func newInt32ValueDecoder(bufferSize int) *int32ValueDecoder {
-	return &int32ValueDecoder{
-		values: make([]int32, 0, atLeastOne(bufferSize/4)),
-	}
-}
-
-func (r *int32ValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueInt32(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeInt32(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *int32ValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *int32ValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type int64ValueDecoder struct {
-	decoder encoding.Decoder
-	values  []int64
-	offset  uint
-}
-
-func newInt64ValueDecoder(bufferSize int) *int64ValueDecoder {
-	return &int64ValueDecoder{
-		values: make([]int64, 0, atLeastOne(bufferSize/8)),
-	}
-}
-
-func (r *int64ValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueInt64(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeInt64(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *int64ValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *int64ValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type int96ValueDecoder struct {
-	decoder encoding.Decoder
-	values  []deprecated.Int96
-	offset  uint
-}
-
-func newInt96ValueDecoder(bufferSize int) *int96ValueDecoder {
-	return &int96ValueDecoder{
-		values: make([]deprecated.Int96, 0, atLeastOne(bufferSize/12)),
-	}
-}
-
-func (r *int96ValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueInt96(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeInt96(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *int96ValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *int96ValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type floatValueDecoder struct {
-	decoder encoding.Decoder
-	values  []float32
-	offset  uint
-}
-
-func newFloatValueDecoder(bufferSize int) *floatValueDecoder {
-	return &floatValueDecoder{
-		values: make([]float32, 0, atLeastOne(bufferSize/4)),
-	}
-}
-
-func (r *floatValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueFloat(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeFloat(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *floatValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *floatValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type doubleValueDecoder struct {
-	decoder encoding.Decoder
-	values  []float64
-	offset  uint
-}
-
-func newDoubleValueDecoder(bufferSize int) *doubleValueDecoder {
-	return &doubleValueDecoder{
-		values: make([]float64, 0, atLeastOne(bufferSize/8)),
-	}
-}
-
-func (r *doubleValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.offset < uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueDouble(r.values[r.offset])
-			r.offset++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeDouble(r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:n]
-		r.offset = 0
-	}
-}
-
-func (r *doubleValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *doubleValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type byteArrayValueDecoder struct {
-	decoder encoding.Decoder
-	values  encoding.ByteArrayList
-	index   int
-}
-
-func newByteArrayValueDecoder(bufferSize int) *byteArrayValueDecoder {
-	return &byteArrayValueDecoder{
-		values: encoding.MakeByteArrayList(atLeastOne(bufferSize / 16)),
-	}
-}
-
-func (r *byteArrayValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for r.index < r.values.Len() && i < len(values) {
-			values[i] = makeValueBytes(ByteArray, copyBytes(r.values.Index(r.index)))
-			r.index++
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		r.values.Reset()
-		n, err := r.decoder.DecodeByteArray(&r.values)
-		if err != nil && n == 0 {
-			return i, err
-		}
-
-		r.index = 0
-	}
-}
-
-func (r *byteArrayValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values.Reset()
-	r.index = 0
-}
-
-func (r *byteArrayValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-type fixedLenByteArrayValueDecoder struct {
-	decoder encoding.Decoder
-	values  []byte
-	offset  uint
-	size    uint
-}
-
-func newFixedLenByteArrayValueDecoder(size, bufferSize int) *fixedLenByteArrayValueDecoder {
-	return &fixedLenByteArrayValueDecoder{
-		size:   uint(size),
-		values: make([]byte, 0, atLeast((bufferSize/size)*size, size)),
-	}
-}
-
-func (r *fixedLenByteArrayValueDecoder) ReadValues(values []Value) (int, error) {
-	i := 0
-	for {
-		for (r.offset+r.size) <= uint(len(r.values)) && i < len(values) {
-			values[i] = makeValueBytes(FixedLenByteArray, copyBytes(r.values[r.offset:r.offset+r.size]))
-			r.offset += r.size
-			i++
-		}
-
-		if i == len(values) {
-			return i, nil
-		}
-		if r.decoder == nil {
-			return i, io.EOF
-		}
-
-		n, err := r.decoder.DecodeFixedLenByteArray(int(r.size), r.values[:cap(r.values)])
-		if n == 0 {
-			return i, err
-		}
-
-		r.values = r.values[:uint(n)*r.size]
-		r.offset = 0
-	}
-}
-
-func (r *fixedLenByteArrayValueDecoder) Reset(decoder encoding.Decoder) {
-	r.decoder = decoder
-	r.values = r.values[:0]
-	r.offset = 0
-}
-
-func (r *fixedLenByteArrayValueDecoder) Decoder() encoding.Decoder {
-	return r.decoder
-}
-
-func atLeastOne(size int) int {
-	return atLeast(size, 1)
-}
-
-func atLeast(size, least int) int {
-	if size < least {
-		return least
-	}
-	return size
-}


Test patch:
diff --git a/bloom_test.go b/bloom_test.go
index 6fd803a..760a832 100644
--- a/bloom_test.go
+++ b/bloom_test.go
@@ -1,8 +1,6 @@
 package parquet
 
 import (
-	"encoding/binary"
-	"math"
 	"math/rand"
 	"testing"
 	"testing/quick"
@@ -20,8 +18,8 @@ func TestBloomFilterEncoder(t *testing.T) {
 		)
 	}
 
-	check := func(e *bloomFilterEncoder, b []byte) bool {
-		return e.filter.Check(e.hash.Sum64(b))
+	check := func(e *bloomFilterEncoder, v Value) bool {
+		return e.filter.Check(v.hash(e.hash))
 	}
 
 	tests := []struct {
@@ -34,43 +32,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -84,9 +46,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -100,9 +60,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -116,11 +74,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -134,9 +88,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -150,9 +102,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				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) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -170,7 +120,7 @@ func TestBloomFilterEncoder(t *testing.T) {
 				f := newFilter(len(values))
 				f.EncodeByteArray(a)
 				for _, v := range values {
-					if !f.Check(v) {
+					if !check(f, ValueOf(v)) {
 						return false
 					}
 				}
@@ -180,15 +130,11 @@ func TestBloomFilterEncoder(t *testing.T) {
 
 		{
 			scenario: "FIXED_LEN_BYTE_ARRAY",
-			function: func(values [][]byte) bool {
-				a := encoding.ByteArrayList{}
-				for _, v := range values {
-					a.Push(v)
-				}
+			function: func(values []byte) bool {
 				f := newFilter(len(values))
-				f.EncodeByteArray(a)
+				f.EncodeFixedLenByteArray(1, values)
 				for _, v := range values {
-					if !f.Check(v) {
+					if !check(f, ValueOf([1]byte{v})) {
 						return false
 					}
 				}
diff --git a/buffer_test.go b/buffer_test.go
index d770f11..b5745c9 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -2,7 +2,6 @@ package parquet_test
 
 import (
 	"bytes"
-	"encoding/binary"
 	"io"
 	"math"
 	"sort"
@@ -198,7 +197,7 @@ func TestBuffer(t *testing.T) {
 									content := new(bytes.Buffer)
 									decoder := parquet.Plain.NewDecoder(content)
 									encoder := parquet.Plain.NewEncoder(content)
-									reader := config.typ.NewValueDecoder(32)
+									reader := config.typ.NewColumnReader(0, 32)
 									buffer := parquet.NewBuffer(options...)
 
 									reset := func() {
@@ -399,9 +398,7 @@ func TestBufferGenerateBloomFilters(t *testing.T) {
 		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 {
+			if ok, err := f.Check(parquet.ValueOf(v)); err != nil {
 				t.Errorf("unexpected error checking bloom filter: %v", err)
 				return false
 			} else if !ok {
diff --git a/page_test.go b/page_test.go
index 036ceff..43eec4f 100644
--- a/page_test.go
+++ b/page_test.go
@@ -1,6 +1,7 @@
 package parquet_test
 
 import (
+	"bytes"
 	"io"
 	"reflect"
 	"testing"
@@ -26,7 +27,7 @@ func testPageBoolean(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value bool }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []bool{false, true}
 				n, err := w.(io.Writer).Write(bits.BoolToBytes(values))
@@ -62,7 +63,7 @@ func testPageInt32(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value int32 }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
 				n, err := w.(io.Writer).Write(bits.Int32ToBytes(values))
@@ -98,7 +99,7 @@ func testPageInt64(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value int64 }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
 				n, err := w.(io.Writer).Write(bits.Int64ToBytes(values))
@@ -134,7 +135,7 @@ func testPageInt96(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value deprecated.Int96 }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []deprecated.Int96{{0: 0}, {0: 1}, {0: 2}}
 				n, err := w.(io.Writer).Write(deprecated.Int96ToBytes(values))
@@ -170,7 +171,7 @@ func testPageFloat(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value float32 }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []float32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
 				n, err := w.(io.Writer).Write(bits.Float32ToBytes(values))
@@ -206,7 +207,7 @@ func testPageDouble(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value float64 }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []float64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
 				n, err := w.(io.Writer).Write(bits.Float64ToBytes(values))
@@ -242,7 +243,7 @@ func testPageByteArray(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value []byte }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []byte{}
 				values = plain.AppendByteArray(values, []byte("A"))
@@ -273,8 +274,8 @@ func testPageByteArray(t *testing.T) {
 
 			read: func(r parquet.ValueReader) (interface{}, error) {
 				values := make([]byte, 3+3*plain.ByteArrayLengthSize)
-				_, err := r.(parquet.ByteArrayReader).ReadByteArrays(values)
-				return values, err
+				n, err := r.(parquet.ByteArrayReader).ReadByteArrays(values)
+				return values[:n+n*plain.ByteArrayLengthSize], err
 			},
 		})
 	})
@@ -284,7 +285,7 @@ func testPageFixedLenByteArray(t *testing.T) {
 	schema := parquet.SchemaOf(struct{ Value [3]byte }{})
 
 	t.Run("io", func(t *testing.T) {
-		testPage(t, schema, pageTest{
+		testBufferPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
 				values := []byte("123456789")
 				n, err := w.(io.Writer).Write(values)
@@ -322,6 +323,11 @@ type pageTest struct {
 }
 
 func testPage(t *testing.T, schema *parquet.Schema, test pageTest) {
+	t.Run("buffer", func(t *testing.T) { testBufferPage(t, schema, test) })
+	t.Run("file", func(t *testing.T) { testFilePage(t, schema, test) })
+}
+
+func testBufferPage(t *testing.T, schema *parquet.Schema, test pageTest) {
 	buffer := parquet.NewBuffer(schema)
 	column := buffer.Column(0).(parquet.ColumnBuffer)
 
@@ -337,5 +343,91 @@ func testPage(t *testing.T, schema *parquet.Schema, test pageTest) {
 	if !reflect.DeepEqual(w, r) {
 		t.Errorf("wrong values read from the page: got=%+v want=%+v", r, w)
 	}
+}
+
+func testFilePage(t *testing.T, schema *parquet.Schema, test pageTest) {
+	buffer := parquet.NewBuffer(schema)
+	column := buffer.Column(0).(parquet.ColumnBuffer)
+
+	w, err := test.write(column)
+	if err != nil {
+		t.Fatal("writing page values:", err)
+	}
+
+	output := new(bytes.Buffer)
+	writer := parquet.NewWriter(output)
+	n, err := writer.WriteRowGroup(buffer)
+	if err != nil {
+		t.Fatal("writing parquet file:", err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal("writing parquet file:", err)
+	}
+	if n != buffer.NumRows() {
+		t.Fatalf("number of rows written mismatch: got=%d want=%d", n, buffer.NumRows())
+	}
+
+	reader := bytes.NewReader(output.Bytes())
+	f, err := parquet.OpenFile(reader, reader.Size())
+	if err != nil {
+		t.Fatal("opening parquet file:", err)
+	}
+
+	p, err := f.RowGroup(0).Column(0).Pages().ReadPage()
+	if err != nil {
+		t.Fatal("reading parquet page:", err)
+	}
+
+	values := p.Values()
+	r, err := test.read(values)
+	if err != io.EOF && err != nil {
+		t.Errorf("expected io.EOF after reading all values but got %v", err)
+	}
+	if !reflect.DeepEqual(w, r) {
+		t.Errorf("wrong values read from the page: got=%+v want=%+v", r, w)
+	}
+	if r, err := test.read(values); reflect.ValueOf(r).Len() != 0 || err != io.EOF {
+		t.Errorf("expected no data and io.EOF after reading all values but got %d and %v", r, err)
+	}
+}
+
+type testStruct struct {
+	Value *string
+}
+
+func TestOptionalPageTrailingNulls(t *testing.T) {
+	schema := parquet.SchemaOf(&testStruct{})
+	buffer := parquet.NewBuffer(schema)
+
+	str := "test"
+	rows := []testStruct{{
+		Value: nil,
+	}, {
+		Value: &str,
+	}, {
+		Value: nil,
+	}}
+
+	for _, row := range rows {
+		if err := buffer.WriteRow(schema.Deconstruct(nil, row)); err != nil {
+			t.Fatal("writing row:", err)
+		}
+	}
 
+	resultRows := []parquet.Row{}
+	reader := buffer.Rows()
+	for {
+		row, err := reader.ReadRow(nil)
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			t.Fatal("reading rows:", err)
+		}
+		resultRows = append(resultRows, row)
+	}
+
+	if len(resultRows) != len(rows) {
+		t.Errorf("wrong number of rows read: got=%d want=%d", len(resultRows), len(rows))
+	}
 }
diff --git a/writer_test.go b/writer_test.go
index 1af09de..b4ab7a2 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -511,7 +511,7 @@ func TestWriterGenerateBloomFilters(t *testing.T) {
 		}
 
 		for i, row := range rows {
-			if ok, err := bloomFilter.Check([]byte(row.LastName)); err != nil {
+			if ok, err := bloomFilter.Check(parquet.ValueOf(row.LastName)); err != nil {
 				t.Errorf("unexpected error checking bloom filter: %v", err)
 				return false
 			} else if !ok {