Repository: segmentio/parquet-go
Pull request #88 (2022-03-07 16:47:00)

Prevent attempting to reorder optional rows when unchanged
In our system, a higher-level layer already ensures that all buffers are sorted, so we do a bunch of unnecessary work attempting to reorder a column whose order hasn't changed. This essentially copies the approach that the repeated buffers took.


Patch:
diff --git a/bloom.go b/bloom.go
index b123e4f..2e50057 100644
--- a/bloom.go
+++ b/bloom.go
@@ -2,12 +2,12 @@ package parquet
 
 import (
 	"io"
-	"unsafe"
 
 	"github.com/segmentio/parquet-go/bloom"
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
+	"github.com/segmentio/parquet-go/internal/bits"
 )
 
 // BloomFilter is an interface allowing applications to test whether a key
@@ -148,23 +148,23 @@ func (e *bloomFilterEncoder) SetBitWidth(int) {
 }
 
 func (e *bloomFilterEncoder) EncodeBoolean(data []bool) error {
-	return e.insert8(*(*[]uint8)(unsafe.Pointer(&data)))
+	return e.insert8(bits.BoolToBytes(data))
 }
 
 func (e *bloomFilterEncoder) EncodeInt8(data []int8) error {
-	return e.insert8(*(*[]uint8)(unsafe.Pointer(&data)))
+	return e.insert8(bits.Int8ToBytes(data))
 }
 
 func (e *bloomFilterEncoder) EncodeInt16(data []int16) error {
-	return e.insert16(*(*[]uint16)(unsafe.Pointer(&data)))
+	return e.insert16(bits.Int16ToUint16(data))
 }
 
 func (e *bloomFilterEncoder) EncodeInt32(data []int32) error {
-	return e.insert32(*(*[]uint32)(unsafe.Pointer(&data)))
+	return e.insert32(bits.Int32ToUint32(data))
 }
 
 func (e *bloomFilterEncoder) EncodeInt64(data []int64) error {
-	return e.insert64(*(*[]uint64)(unsafe.Pointer(&data)))
+	return e.insert64(bits.Int64ToUint64(data))
 }
 
 func (e *bloomFilterEncoder) EncodeInt96(data []deprecated.Int96) error {
@@ -172,11 +172,11 @@ func (e *bloomFilterEncoder) EncodeInt96(data []deprecated.Int96) error {
 }
 
 func (e *bloomFilterEncoder) EncodeFloat(data []float32) error {
-	return e.insert32(*(*[]uint32)(unsafe.Pointer(&data)))
+	return e.insert32(bits.Float32ToUint32(data))
 }
 
 func (e *bloomFilterEncoder) EncodeDouble(data []float64) error {
-	return e.insert64(*(*[]uint64)(unsafe.Pointer(&data)))
+	return e.insert64(bits.Float64ToUint64(data))
 }
 
 func (e *bloomFilterEncoder) EncodeByteArray(data encoding.ByteArrayList) error {
@@ -186,7 +186,7 @@ func (e *bloomFilterEncoder) EncodeByteArray(data encoding.ByteArrayList) error
 
 func (e *bloomFilterEncoder) EncodeFixedLenByteArray(size int, data []byte) error {
 	if size == 16 {
-		return e.insert128(*(*[][16]byte)(unsafe.Pointer(&data)))
+		return e.insert128(bits.BytesToUint128(data))
 	}
 	for i, j := 0, size; j <= len(data); {
 		e.insert(data[i:j])
diff --git a/class_go18.go b/class_go18.go
new file mode 100644
index 0000000..052731b
--- /dev/null
+++ b/class_go18.go
@@ -0,0 +1,184 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"unsafe"
+
+	"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"
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+type primitive interface {
+	bool | int32 | int64 | deprecated.Int96 | float32 | float64 | uint32 | uint64
+}
+
+func sizeof[T primitive]() int {
+	var z T
+	return int(unsafe.Sizeof(z))
+}
+
+type comparable interface {
+	int32 | int64 | float32 | float64 | uint32 | uint64
+}
+
+func compare[T comparable](a, b T) int {
+	switch {
+	case a < b:
+		return -1
+	case a > b:
+		return +1
+	default:
+		return 0
+	}
+}
+
+func less[T comparable](a, b T) bool {
+	return a < b
+}
+
+type class[T primitive] struct {
+	name      string
+	makeValue func(T) Value
+	value     func(Value) T
+	plain     func(T) []byte
+	compare   func(T, T) int
+	less      func(T, T) bool
+	order     func([]T) int
+	min       func([]T) T
+	max       func([]T) T
+	bounds    func([]T) (T, T)
+	encode    func(encoding.Encoder, []T) error
+	decode    func(encoding.Decoder, []T) (int, error)
+}
+
+var boolClass = class[bool]{
+	name:      "BOOLEAN",
+	makeValue: makeValueBoolean,
+	value:     Value.Boolean,
+	plain:     plain.Boolean,
+	compare:   compareBool,
+	less:      func(a, b bool) bool { return a != b && !a },
+	order:     bits.OrderOfBool,
+	min:       bits.MinBool,
+	max:       bits.MaxBool,
+	bounds:    bits.MinMaxBool,
+	encode:    encoding.Encoder.EncodeBoolean,
+	decode:    encoding.Decoder.DecodeBoolean,
+}
+
+var int32Class = class[int32]{
+	name:      "INT32",
+	makeValue: makeValueInt32,
+	value:     Value.Int32,
+	plain:     plain.Int32,
+	compare:   compare[int32],
+	less:      less[int32],
+	order:     bits.OrderOfInt32,
+	min:       bits.MinInt32,
+	max:       bits.MaxInt32,
+	bounds:    bits.MinMaxInt32,
+	encode:    encoding.Encoder.EncodeInt32,
+	decode:    encoding.Decoder.DecodeInt32,
+}
+
+var int64Class = class[int64]{
+	name:      "INT64",
+	makeValue: makeValueInt64,
+	value:     Value.Int64,
+	plain:     plain.Int64,
+	compare:   compare[int64],
+	less:      less[int64],
+	order:     bits.OrderOfInt64,
+	min:       bits.MinInt64,
+	max:       bits.MaxInt64,
+	bounds:    bits.MinMaxInt64,
+	encode:    encoding.Encoder.EncodeInt64,
+	decode:    encoding.Decoder.DecodeInt64,
+}
+
+var int96Class = class[deprecated.Int96]{
+	name:      "INT96",
+	makeValue: makeValueInt96,
+	value:     Value.Int96,
+	plain:     plain.Int96,
+	compare:   compareInt96,
+	less:      deprecated.Int96.Less,
+	order:     deprecated.OrderOfInt96,
+	min:       deprecated.MinInt96,
+	max:       deprecated.MaxInt96,
+	bounds:    deprecated.MinMaxInt96,
+	encode:    encoding.Encoder.EncodeInt96,
+	decode:    encoding.Decoder.DecodeInt96,
+}
+
+var float32Class = class[float32]{
+	name:      "FLOAT",
+	makeValue: makeValueFloat,
+	value:     Value.Float,
+	plain:     plain.Float,
+	compare:   compare[float32],
+	less:      less[float32],
+	order:     bits.OrderOfFloat32,
+	min:       bits.MinFloat32,
+	max:       bits.MaxFloat32,
+	bounds:    bits.MinMaxFloat32,
+	encode:    encoding.Encoder.EncodeFloat,
+	decode:    encoding.Decoder.DecodeFloat,
+}
+
+var float64Class = class[float64]{
+	name:      "DOUBLE",
+	makeValue: makeValueDouble,
+	value:     Value.Double,
+	plain:     plain.Double,
+	compare:   compare[float64],
+	less:      less[float64],
+	order:     bits.OrderOfFloat64,
+	min:       bits.MinFloat64,
+	max:       bits.MaxFloat64,
+	bounds:    bits.MinMaxFloat64,
+	encode:    encoding.Encoder.EncodeDouble,
+	decode:    encoding.Decoder.DecodeDouble,
+}
+
+var uint32Class = class[uint32]{
+	name:      "INT32",
+	makeValue: func(v uint32) Value { return makeValueInt32(int32(v)) },
+	value:     func(v Value) uint32 { return uint32(v.Int32()) },
+	plain:     func(v uint32) []byte { return plain.Int32(int32(v)) },
+	compare:   compare[uint32],
+	less:      less[uint32],
+	order:     bits.OrderOfUint32,
+	min:       bits.MinUint32,
+	max:       bits.MaxUint32,
+	bounds:    bits.MinMaxUint32,
+	encode: func(e encoding.Encoder, v []uint32) error {
+		return e.EncodeInt32(cast.Slice[int32](v))
+	},
+	decode: func(d encoding.Decoder, v []uint32) (int, error) {
+		return d.DecodeInt32(cast.Slice[int32](v))
+	},
+}
+
+var uint64Class = class[uint64]{
+	name:      "INT64",
+	makeValue: func(v uint64) Value { return makeValueInt64(int64(v)) },
+	value:     func(v Value) uint64 { return uint64(v.Int64()) },
+	plain:     func(v uint64) []byte { return plain.Int64(int64(v)) },
+	compare:   compare[uint64],
+	less:      less[uint64],
+	order:     bits.OrderOfUint64,
+	min:       bits.MinUint64,
+	max:       bits.MaxUint64,
+	bounds:    bits.MinMaxUint64,
+	encode: func(e encoding.Encoder, v []uint64) error {
+		return e.EncodeInt64(cast.Slice[int64](v))
+	},
+	decode: func(d encoding.Decoder, v []uint64) (int, error) {
+		return d.DecodeInt64(cast.Slice[int64](v))
+	},
+}
diff --git a/column_buffer.go b/column_buffer.go
index 6b96eaa..d0d419f 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -6,10 +6,8 @@ import (
 	"io"
 	"sort"
 
-	"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"
 )
 
 // ColumnBuffer is an interface representing columns of a row group.
@@ -184,6 +182,13 @@ func (col *optionalColumnBuffer) Pages() Pages {
 }
 
 func (col *optionalColumnBuffer) Page() BufferedPage {
+	if !optionalRowsHaveBeenReordered(col.rows) {
+		// No need for any cyclic sorting if the rows have not been reordered.
+		// This case is also important because the cyclic sorting modifies the
+		// buffer which makes it unsafe to read the buffer concurrently.
+		return newOptionalPage(col.base.Page(), col.maxDefinitionLevel, col.definitionLevels)
+	}
+
 	numNulls := countLevelsNotEqual(col.definitionLevels, col.maxDefinitionLevel)
 	numValues := len(col.rows) - numNulls
 
@@ -434,7 +439,7 @@ func (col *repeatedColumnBuffer) Pages() Pages {
 }
 
 func (col *repeatedColumnBuffer) Page() BufferedPage {
-	if rowsHaveBeenReordered(col.rows) {
+	if repeatedRowsHaveBeenReordered(col.rows) {
 		if col.reordering == nil {
 			col.reordering = col.Clone().(*repeatedColumnBuffer)
 		}
@@ -644,7 +649,27 @@ func (col *repeatedColumnBuffer) Values() ValueReader {
 	return &repeatedPageReader{page: col.Page().(*repeatedPage)}
 }
 
-func rowsHaveBeenReordered(rows []region) bool {
+func optionalRowsHaveBeenReordered(rows []int32) bool {
+	i := int32(0)
+	for _, row := range rows {
+		if row < 0 {
+			// Skip any row that is null.
+			continue
+		}
+
+		// If rows have been reordered the indices are not increasing exactly
+		// one by one.
+		if row != i {
+			return true
+		}
+
+		// Only increment the index if the row is not null.
+		i++
+	}
+	return false
+}
+
+func repeatedRowsHaveBeenReordered(rows []region) bool {
 	offset := uint32(0)
 	for _, row := range rows {
 		if row.offset != offset {
@@ -664,586 +689,6 @@ func maxRowLengthOf(rows []region) (maxLength uint32) {
 	return maxLength
 }
 
-// =============================================================================
-// The types below are in-memory implementations of the ColumnBuffer interface
-// for each parquet type.
-//
-// These column buffers are created by calling NewColumnBuffer on parquet.Type
-// instances; each parquet type manages to construct column buffers of the
-// appropriate type, which ensures that we are packing as many values as we
-// can in memory.
-//
-// See Type.NewColumnBuffer for details about how these types get created.
-// =============================================================================
-
-type booleanColumnBuffer struct {
-	booleanPage
-	typ Type
-}
-
-func newBooleanColumnBuffer(typ Type, columnIndex int16, bufferSize int) *booleanColumnBuffer {
-	return &booleanColumnBuffer{
-		booleanPage: booleanPage{
-			values:      make([]bool, 0, bufferSize),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *booleanColumnBuffer) Clone() ColumnBuffer {
-	return &booleanColumnBuffer{
-		booleanPage: booleanPage{
-			values:      append([]bool{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *booleanColumnBuffer) Type() Type { return col.typ }
-
-func (col *booleanColumnBuffer) ColumnIndex() ColumnIndex { return booleanPageIndex{&col.booleanPage} }
-
-func (col *booleanColumnBuffer) OffsetIndex() OffsetIndex { return booleanPageIndex{&col.booleanPage} }
-
-func (col *booleanColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *booleanColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *booleanColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *booleanColumnBuffer) Page() BufferedPage { return &col.booleanPage }
-
-func (col *booleanColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *booleanColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *booleanColumnBuffer) Len() int { return len(col.values) }
-
-func (col *booleanColumnBuffer) Less(i, j int) bool {
-	return col.values[i] != col.values[j] && !col.values[i]
-}
-
-func (col *booleanColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *booleanColumnBuffer) Write(b []byte) (int, error) {
-	return col.WriteBooleans(bits.BytesToBool(b))
-}
-
-func (col *booleanColumnBuffer) WriteBooleans(values []bool) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Boolean())
-	}
-	return len(values), nil
-}
-
-func (col *booleanColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Boolean())
-	return nil
-}
-
-func (col *booleanColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueBoolean(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
-type int32ColumnBuffer struct {
-	int32Page
-	typ Type
-}
-
-func newInt32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int32ColumnBuffer {
-	return &int32ColumnBuffer{
-		int32Page: int32Page{
-			values:      make([]int32, 0, bufferSize/4),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *int32ColumnBuffer) Clone() ColumnBuffer {
-	return &int32ColumnBuffer{
-		int32Page: int32Page{
-			values:      append([]int32{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *int32ColumnBuffer) Type() Type { return col.typ }
-
-func (col *int32ColumnBuffer) ColumnIndex() ColumnIndex { return int32PageIndex{&col.int32Page} }
-
-func (col *int32ColumnBuffer) OffsetIndex() OffsetIndex { return int32PageIndex{&col.int32Page} }
-
-func (col *int32ColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *int32ColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *int32ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *int32ColumnBuffer) Page() BufferedPage { return &col.int32Page }
-
-func (col *int32ColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *int32ColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *int32ColumnBuffer) Len() int { return len(col.values) }
-
-func (col *int32ColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
-
-func (col *int32ColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *int32ColumnBuffer) Write(b []byte) (int, error) {
-	if (len(b) % 4) != 0 {
-		return 0, fmt.Errorf("cannot write INT32 values from input of size %d", len(b))
-	}
-	col.values = append(col.values, bits.BytesToInt32(b)...)
-	return len(b), nil
-}
-
-func (col *int32ColumnBuffer) WriteInt32s(values []int32) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Int32())
-	}
-	return len(values), nil
-}
-
-func (col *int32ColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Int32())
-	return nil
-}
-
-func (col *int32ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueInt32(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
-type int64ColumnBuffer struct {
-	int64Page
-	typ Type
-}
-
-func newInt64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int64ColumnBuffer {
-	return &int64ColumnBuffer{
-		int64Page: int64Page{
-			values:      make([]int64, 0, bufferSize/8),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *int64ColumnBuffer) Clone() ColumnBuffer {
-	return &int64ColumnBuffer{
-		int64Page: int64Page{
-			values:      append([]int64{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *int64ColumnBuffer) Type() Type { return col.typ }
-
-func (col *int64ColumnBuffer) ColumnIndex() ColumnIndex { return int64PageIndex{&col.int64Page} }
-
-func (col *int64ColumnBuffer) OffsetIndex() OffsetIndex { return int64PageIndex{&col.int64Page} }
-
-func (col *int64ColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *int64ColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *int64ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *int64ColumnBuffer) Page() BufferedPage { return &col.int64Page }
-
-func (col *int64ColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *int64ColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *int64ColumnBuffer) Len() int { return len(col.values) }
-
-func (col *int64ColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
-
-func (col *int64ColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *int64ColumnBuffer) Write(b []byte) (int, error) {
-	if (len(b) % 8) != 0 {
-		return 0, fmt.Errorf("cannot write INT64 values from input of size %d", len(b))
-	}
-	col.values = append(col.values, bits.BytesToInt64(b)...)
-	return len(b), nil
-}
-
-func (col *int64ColumnBuffer) WriteInt64s(values []int64) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Int64())
-	}
-	return len(values), nil
-}
-
-func (col *int64ColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Int64())
-	return nil
-}
-
-func (col *int64ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueInt64(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
-type int96ColumnBuffer struct {
-	int96Page
-	typ Type
-}
-
-func newInt96ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int96ColumnBuffer {
-	return &int96ColumnBuffer{
-		int96Page: int96Page{
-			values:      make([]deprecated.Int96, 0, bufferSize/12),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *int96ColumnBuffer) Clone() ColumnBuffer {
-	return &int96ColumnBuffer{
-		int96Page: int96Page{
-			values:      append([]deprecated.Int96{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *int96ColumnBuffer) Type() Type { return col.typ }
-
-func (col *int96ColumnBuffer) ColumnIndex() ColumnIndex { return int96PageIndex{&col.int96Page} }
-
-func (col *int96ColumnBuffer) OffsetIndex() OffsetIndex { return int96PageIndex{&col.int96Page} }
-
-func (col *int96ColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *int96ColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *int96ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *int96ColumnBuffer) Page() BufferedPage { return &col.int96Page }
-
-func (col *int96ColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *int96ColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *int96ColumnBuffer) Len() int { return len(col.values) }
-
-func (col *int96ColumnBuffer) Less(i, j int) bool { return col.values[i].Less(col.values[j]) }
-
-func (col *int96ColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *int96ColumnBuffer) Write(b []byte) (int, error) {
-	if (len(b) % 12) != 0 {
-		return 0, fmt.Errorf("cannot write INT96 values from input of size %d", len(b))
-	}
-	col.values = append(col.values, deprecated.BytesToInt96(b)...)
-	return len(b), nil
-}
-
-func (col *int96ColumnBuffer) WriteInt96s(values []deprecated.Int96) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Int96())
-	}
-	return len(values), nil
-}
-
-func (col *int96ColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Int96())
-	return nil
-}
-
-func (col *int96ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueInt96(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
-type floatColumnBuffer struct {
-	floatPage
-	typ Type
-}
-
-func newFloatColumnBuffer(typ Type, columnIndex int16, bufferSize int) *floatColumnBuffer {
-	return &floatColumnBuffer{
-		floatPage: floatPage{
-			values:      make([]float32, 0, bufferSize/4),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *floatColumnBuffer) Clone() ColumnBuffer {
-	return &floatColumnBuffer{
-		floatPage: floatPage{
-			values:      append([]float32{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *floatColumnBuffer) Type() Type { return col.typ }
-
-func (col *floatColumnBuffer) ColumnIndex() ColumnIndex { return floatPageIndex{&col.floatPage} }
-
-func (col *floatColumnBuffer) OffsetIndex() OffsetIndex { return floatPageIndex{&col.floatPage} }
-
-func (col *floatColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *floatColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *floatColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *floatColumnBuffer) Page() BufferedPage { return &col.floatPage }
-
-func (col *floatColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *floatColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *floatColumnBuffer) Len() int { return len(col.values) }
-
-func (col *floatColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
-
-func (col *floatColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *floatColumnBuffer) Write(b []byte) (int, error) {
-	if (len(b) % 4) != 0 {
-		return 0, fmt.Errorf("cannot write FLOAT values from input of size %d", len(b))
-	}
-	col.values = append(col.values, bits.BytesToFloat32(b)...)
-	return len(b), nil
-}
-
-func (col *floatColumnBuffer) WriteFloats(values []float32) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Float())
-	}
-	return len(values), nil
-}
-
-func (col *floatColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Float())
-	return nil
-}
-
-func (col *floatColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueFloat(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
-type doubleColumnBuffer struct {
-	doublePage
-	typ Type
-}
-
-func newDoubleColumnBuffer(typ Type, columnIndex int16, bufferSize int) *doubleColumnBuffer {
-	return &doubleColumnBuffer{
-		doublePage: doublePage{
-			values:      make([]float64, 0, bufferSize/8),
-			columnIndex: ^columnIndex,
-		},
-		typ: typ,
-	}
-}
-
-func (col *doubleColumnBuffer) Clone() ColumnBuffer {
-	return &doubleColumnBuffer{
-		doublePage: doublePage{
-			values:      append([]float64{}, col.values...),
-			columnIndex: col.columnIndex,
-		},
-		typ: col.typ,
-	}
-}
-
-func (col *doubleColumnBuffer) Type() Type { return col.typ }
-
-func (col *doubleColumnBuffer) ColumnIndex() ColumnIndex { return doublePageIndex{&col.doublePage} }
-
-func (col *doubleColumnBuffer) OffsetIndex() OffsetIndex { return doublePageIndex{&col.doublePage} }
-
-func (col *doubleColumnBuffer) BloomFilter() BloomFilter { return nil }
-
-func (col *doubleColumnBuffer) Dictionary() Dictionary { return nil }
-
-func (col *doubleColumnBuffer) Pages() Pages { return onePage(col.Page()) }
-
-func (col *doubleColumnBuffer) Page() BufferedPage { return &col.doublePage }
-
-func (col *doubleColumnBuffer) Reset() { col.values = col.values[:0] }
-
-func (col *doubleColumnBuffer) Cap() int { return cap(col.values) }
-
-func (col *doubleColumnBuffer) Len() int { return len(col.values) }
-
-func (col *doubleColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
-
-func (col *doubleColumnBuffer) Swap(i, j int) {
-	col.values[i], col.values[j] = col.values[j], col.values[i]
-}
-
-func (col *doubleColumnBuffer) Write(b []byte) (int, error) {
-	if (len(b) % 8) != 0 {
-		return 0, fmt.Errorf("cannot write DOUBLE values from input of size %d", len(b))
-	}
-	col.values = append(col.values, bits.BytesToFloat64(b)...)
-	return len(b), nil
-}
-
-func (col *doubleColumnBuffer) WriteDoubles(values []float64) (int, error) {
-	col.values = append(col.values, values...)
-	return len(values), nil
-}
-
-func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, v.Double())
-	}
-	return len(values), nil
-}
-
-func (col *doubleColumnBuffer) WriteRow(row Row) error {
-	if len(row) == 0 {
-		return errRowHasTooFewValues(int64(len(row)))
-	}
-	if len(row) > 1 {
-		return errRowHasTooManyValues(int64(len(row)))
-	}
-	col.values = append(col.values, row[0].Double())
-	return nil
-}
-
-func (col *doubleColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
-	switch {
-	case index < 0:
-		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
-	case index >= int64(len(col.values)):
-		return row, io.EOF
-	default:
-		v := makeValueDouble(col.values[index])
-		v.columnIndex = col.columnIndex
-		return append(row, v), nil
-	}
-}
-
 type byteArrayColumnBuffer struct {
 	byteArrayPage
 	typ Type
@@ -1272,11 +717,11 @@ func (col *byteArrayColumnBuffer) Clone() ColumnBuffer {
 func (col *byteArrayColumnBuffer) Type() Type { return col.typ }
 
 func (col *byteArrayColumnBuffer) ColumnIndex() ColumnIndex {
-	return byteArrayPageIndex{&col.byteArrayPage}
+	return byteArrayColumnIndex{&col.byteArrayPage}
 }
 
 func (col *byteArrayColumnBuffer) OffsetIndex() OffsetIndex {
-	return byteArrayPageIndex{&col.byteArrayPage}
+	return byteArrayOffsetIndex{&col.byteArrayPage}
 }
 
 func (col *byteArrayColumnBuffer) BloomFilter() BloomFilter { return nil }
@@ -1302,6 +747,10 @@ func (col *byteArrayColumnBuffer) Write(b []byte) (int, error) {
 	return n, err
 }
 
+func (col *byteArrayColumnBuffer) WriteRequired(values []byte) (int, error) {
+	return col.WriteByteArrays(values)
+}
+
 func (col *byteArrayColumnBuffer) WriteByteArrays(values []byte) (int, error) {
 	n, _, err := col.writeByteArrays(values)
 	return n, err
@@ -1382,11 +831,11 @@ func (col *fixedLenByteArrayColumnBuffer) Clone() ColumnBuffer {
 func (col *fixedLenByteArrayColumnBuffer) Type() Type { return col.typ }
 
 func (col *fixedLenByteArrayColumnBuffer) ColumnIndex() ColumnIndex {
-	return fixedLenByteArrayPageIndex{&col.fixedLenByteArrayPage}
+	return fixedLenByteArrayColumnIndex{&col.fixedLenByteArrayPage}
 }
 
 func (col *fixedLenByteArrayColumnBuffer) OffsetIndex() OffsetIndex {
-	return fixedLenByteArrayPageIndex{&col.fixedLenByteArrayPage}
+	return fixedLenByteArrayOffsetIndex{&col.fixedLenByteArrayPage}
 }
 
 func (col *fixedLenByteArrayColumnBuffer) BloomFilter() BloomFilter { return nil }
@@ -1425,6 +874,10 @@ func (col *fixedLenByteArrayColumnBuffer) Write(b []byte) (int, error) {
 	return n * col.size, err
 }
 
+func (col *fixedLenByteArrayColumnBuffer) WriteRequired(values []byte) (int, error) {
+	return col.WriteFixedLenByteArrays(values)
+}
+
 func (col *fixedLenByteArrayColumnBuffer) WriteFixedLenByteArrays(values []byte) (int, error) {
 	d, m := len(values)/col.size, len(values)%col.size
 	if m != 0 {
@@ -1467,72 +920,8 @@ func (col *fixedLenByteArrayColumnBuffer) ReadRowAt(row Row, index int64) (Row,
 	}
 }
 
-type uint32ColumnBuffer struct{ *int32ColumnBuffer }
-
-func newUint32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint32ColumnBuffer {
-	return uint32ColumnBuffer{newInt32ColumnBuffer(typ, columnIndex, bufferSize)}
-}
-
-func (col uint32ColumnBuffer) ColumnIndex() ColumnIndex { return uint32PageIndex{col.page()} }
-
-func (col uint32ColumnBuffer) OffsetIndex() OffsetIndex { return uint32PageIndex{col.page()} }
-
-func (col uint32ColumnBuffer) page() uint32Page { return uint32Page{&col.int32Page} }
-
-func (col uint32ColumnBuffer) Page() BufferedPage { return col.page() }
-
-func (col uint32ColumnBuffer) Pages() Pages { return onePage(col.page()) }
-
-func (col uint32ColumnBuffer) Clone() ColumnBuffer {
-	return uint32ColumnBuffer{col.int32ColumnBuffer.Clone().(*int32ColumnBuffer)}
-}
-
-func (col uint32ColumnBuffer) Less(i, j int) bool {
-	return uint32(col.values[i]) < uint32(col.values[j])
-}
-
-type uint64ColumnBuffer struct{ *int64ColumnBuffer }
-
-func newUint64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint64ColumnBuffer {
-	return uint64ColumnBuffer{newInt64ColumnBuffer(typ, columnIndex, bufferSize)}
-}
-
-func (col uint64ColumnBuffer) ColumnIndex() ColumnIndex { return uint64PageIndex{col.page()} }
-
-func (col uint64ColumnBuffer) OffsetIndex() OffsetIndex { return uint64PageIndex{col.page()} }
-
-func (col uint64ColumnBuffer) page() uint64Page { return uint64Page{&col.int64Page} }
-
-func (col uint64ColumnBuffer) Page() BufferedPage { return col.page() }
-
-func (col uint64ColumnBuffer) Pages() Pages { return onePage(col.page()) }
-
-func (col uint64ColumnBuffer) Clone() ColumnBuffer {
-	return uint64ColumnBuffer{col.int64ColumnBuffer.Clone().(*int64ColumnBuffer)}
-}
-
-func (col uint64ColumnBuffer) Less(i, j int) bool {
-	return uint64(col.values[i]) < uint64(col.values[j])
-}
-
 var (
 	_ sort.Interface = (ColumnBuffer)(nil)
-
-	_ io.Writer = (*booleanColumnBuffer)(nil)
-	_ io.Writer = (*int32ColumnBuffer)(nil)
-	_ io.Writer = (*int64ColumnBuffer)(nil)
-	_ io.Writer = (*int96ColumnBuffer)(nil)
-	_ io.Writer = (*floatColumnBuffer)(nil)
-	_ io.Writer = (*doubleColumnBuffer)(nil)
-	_ io.Writer = (*byteArrayColumnBuffer)(nil)
-	_ io.Writer = (*fixedLenByteArrayColumnBuffer)(nil)
-
-	_ BooleanWriter           = (*booleanColumnBuffer)(nil)
-	_ Int32Writer             = (*int32ColumnBuffer)(nil)
-	_ Int64Writer             = (*int64ColumnBuffer)(nil)
-	_ Int96Writer             = (*int96ColumnBuffer)(nil)
-	_ FloatWriter             = (*floatColumnBuffer)(nil)
-	_ DoubleWriter            = (*doubleColumnBuffer)(nil)
-	_ ByteArrayWriter         = (*byteArrayColumnBuffer)(nil)
-	_ FixedLenByteArrayWriter = (*fixedLenByteArrayColumnBuffer)(nil)
+	_ io.Writer      = (*byteArrayColumnBuffer)(nil)
+	_ io.Writer      = (*fixedLenByteArrayColumnBuffer)(nil)
 )
diff --git a/column_buffer_default.go b/column_buffer_default.go
new file mode 100644
index 0000000..1d390bd
--- /dev/null
+++ b/column_buffer_default.go
@@ -0,0 +1,661 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+// =============================================================================
+// The types below are in-memory implementations of the ColumnBuffer interface
+// for each parquet type.
+//
+// These column buffers are created by calling NewColumnBuffer on parquet.Type
+// instances; each parquet type manages to construct column buffers of the
+// appropriate type, which ensures that we are packing as many values as we
+// can in memory.
+//
+// See Type.NewColumnBuffer for details about how these types get created.
+// =============================================================================
+
+type booleanColumnBuffer struct {
+	booleanPage
+	typ Type
+}
+
+func newBooleanColumnBuffer(typ Type, columnIndex int16, bufferSize int) *booleanColumnBuffer {
+	return &booleanColumnBuffer{
+		booleanPage: booleanPage{
+			values:      make([]bool, 0, bufferSize),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *booleanColumnBuffer) Clone() ColumnBuffer {
+	return &booleanColumnBuffer{
+		booleanPage: booleanPage{
+			values:      append([]bool{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *booleanColumnBuffer) Type() Type { return col.typ }
+
+func (col *booleanColumnBuffer) ColumnIndex() ColumnIndex {
+	return booleanColumnIndex{&col.booleanPage}
+}
+
+func (col *booleanColumnBuffer) OffsetIndex() OffsetIndex {
+	return booleanOffsetIndex{&col.booleanPage}
+}
+
+func (col *booleanColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *booleanColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *booleanColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *booleanColumnBuffer) Page() BufferedPage { return &col.booleanPage }
+
+func (col *booleanColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *booleanColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *booleanColumnBuffer) Len() int { return len(col.values) }
+
+func (col *booleanColumnBuffer) Less(i, j int) bool {
+	return col.values[i] != col.values[j] && !col.values[i]
+}
+
+func (col *booleanColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *booleanColumnBuffer) Write(b []byte) (int, error) {
+	return col.WriteBooleans(bits.BytesToBool(b))
+}
+
+func (col *booleanColumnBuffer) WriteBooleans(values []bool) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Boolean())
+	}
+	return len(values), nil
+}
+
+func (col *booleanColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Boolean())
+	return nil
+}
+
+func (col *booleanColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueBoolean(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type int32ColumnBuffer struct {
+	int32Page
+	typ Type
+}
+
+func newInt32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int32ColumnBuffer {
+	return &int32ColumnBuffer{
+		int32Page: int32Page{
+			values:      make([]int32, 0, bufferSize/4),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *int32ColumnBuffer) Clone() ColumnBuffer {
+	return &int32ColumnBuffer{
+		int32Page: int32Page{
+			values:      append([]int32{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *int32ColumnBuffer) Type() Type { return col.typ }
+
+func (col *int32ColumnBuffer) ColumnIndex() ColumnIndex { return int32ColumnIndex{&col.int32Page} }
+
+func (col *int32ColumnBuffer) OffsetIndex() OffsetIndex { return int32OffsetIndex{&col.int32Page} }
+
+func (col *int32ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *int32ColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *int32ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *int32ColumnBuffer) Page() BufferedPage { return &col.int32Page }
+
+func (col *int32ColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *int32ColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *int32ColumnBuffer) Len() int { return len(col.values) }
+
+func (col *int32ColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
+
+func (col *int32ColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *int32ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 4) != 0 {
+		return 0, fmt.Errorf("cannot write INT32 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToInt32(b)...)
+	return len(b), nil
+}
+
+func (col *int32ColumnBuffer) WriteInt32s(values []int32) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Int32())
+	}
+	return len(values), nil
+}
+
+func (col *int32ColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Int32())
+	return nil
+}
+
+func (col *int32ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueInt32(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type int64ColumnBuffer struct {
+	int64Page
+	typ Type
+}
+
+func newInt64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int64ColumnBuffer {
+	return &int64ColumnBuffer{
+		int64Page: int64Page{
+			values:      make([]int64, 0, bufferSize/8),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *int64ColumnBuffer) Clone() ColumnBuffer {
+	return &int64ColumnBuffer{
+		int64Page: int64Page{
+			values:      append([]int64{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *int64ColumnBuffer) Type() Type { return col.typ }
+
+func (col *int64ColumnBuffer) ColumnIndex() ColumnIndex { return int64ColumnIndex{&col.int64Page} }
+
+func (col *int64ColumnBuffer) OffsetIndex() OffsetIndex { return int64OffsetIndex{&col.int64Page} }
+
+func (col *int64ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *int64ColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *int64ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *int64ColumnBuffer) Page() BufferedPage { return &col.int64Page }
+
+func (col *int64ColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *int64ColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *int64ColumnBuffer) Len() int { return len(col.values) }
+
+func (col *int64ColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
+
+func (col *int64ColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *int64ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 8) != 0 {
+		return 0, fmt.Errorf("cannot write INT64 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToInt64(b)...)
+	return len(b), nil
+}
+
+func (col *int64ColumnBuffer) WriteInt64s(values []int64) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Int64())
+	}
+	return len(values), nil
+}
+
+func (col *int64ColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Int64())
+	return nil
+}
+
+func (col *int64ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueInt64(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type int96ColumnBuffer struct {
+	int96Page
+	typ Type
+}
+
+func newInt96ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *int96ColumnBuffer {
+	return &int96ColumnBuffer{
+		int96Page: int96Page{
+			values:      make([]deprecated.Int96, 0, bufferSize/12),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *int96ColumnBuffer) Clone() ColumnBuffer {
+	return &int96ColumnBuffer{
+		int96Page: int96Page{
+			values:      append([]deprecated.Int96{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *int96ColumnBuffer) Type() Type { return col.typ }
+
+func (col *int96ColumnBuffer) ColumnIndex() ColumnIndex { return int96ColumnIndex{&col.int96Page} }
+
+func (col *int96ColumnBuffer) OffsetIndex() OffsetIndex { return int96OffsetIndex{&col.int96Page} }
+
+func (col *int96ColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *int96ColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *int96ColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *int96ColumnBuffer) Page() BufferedPage { return &col.int96Page }
+
+func (col *int96ColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *int96ColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *int96ColumnBuffer) Len() int { return len(col.values) }
+
+func (col *int96ColumnBuffer) Less(i, j int) bool { return col.values[i].Less(col.values[j]) }
+
+func (col *int96ColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *int96ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 12) != 0 {
+		return 0, fmt.Errorf("cannot write INT96 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, deprecated.BytesToInt96(b)...)
+	return len(b), nil
+}
+
+func (col *int96ColumnBuffer) WriteInt96s(values []deprecated.Int96) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Int96())
+	}
+	return len(values), nil
+}
+
+func (col *int96ColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Int96())
+	return nil
+}
+
+func (col *int96ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueInt96(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type floatColumnBuffer struct {
+	floatPage
+	typ Type
+}
+
+func newFloatColumnBuffer(typ Type, columnIndex int16, bufferSize int) *floatColumnBuffer {
+	return &floatColumnBuffer{
+		floatPage: floatPage{
+			values:      make([]float32, 0, bufferSize/4),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *floatColumnBuffer) Clone() ColumnBuffer {
+	return &floatColumnBuffer{
+		floatPage: floatPage{
+			values:      append([]float32{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *floatColumnBuffer) Type() Type { return col.typ }
+
+func (col *floatColumnBuffer) ColumnIndex() ColumnIndex { return floatColumnIndex{&col.floatPage} }
+
+func (col *floatColumnBuffer) OffsetIndex() OffsetIndex { return floatOffsetIndex{&col.floatPage} }
+
+func (col *floatColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *floatColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *floatColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *floatColumnBuffer) Page() BufferedPage { return &col.floatPage }
+
+func (col *floatColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *floatColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *floatColumnBuffer) Len() int { return len(col.values) }
+
+func (col *floatColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
+
+func (col *floatColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *floatColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 4) != 0 {
+		return 0, fmt.Errorf("cannot write FLOAT values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToFloat32(b)...)
+	return len(b), nil
+}
+
+func (col *floatColumnBuffer) WriteFloats(values []float32) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Float())
+	}
+	return len(values), nil
+}
+
+func (col *floatColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Float())
+	return nil
+}
+
+func (col *floatColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueFloat(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type doubleColumnBuffer struct {
+	doublePage
+	typ Type
+}
+
+func newDoubleColumnBuffer(typ Type, columnIndex int16, bufferSize int) *doubleColumnBuffer {
+	return &doubleColumnBuffer{
+		doublePage: doublePage{
+			values:      make([]float64, 0, bufferSize/8),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *doubleColumnBuffer) Clone() ColumnBuffer {
+	return &doubleColumnBuffer{
+		doublePage: doublePage{
+			values:      append([]float64{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *doubleColumnBuffer) Type() Type { return col.typ }
+
+func (col *doubleColumnBuffer) ColumnIndex() ColumnIndex { return doubleColumnIndex{&col.doublePage} }
+
+func (col *doubleColumnBuffer) OffsetIndex() OffsetIndex { return doubleOffsetIndex{&col.doublePage} }
+
+func (col *doubleColumnBuffer) BloomFilter() BloomFilter { return nil }
+
+func (col *doubleColumnBuffer) Dictionary() Dictionary { return nil }
+
+func (col *doubleColumnBuffer) Pages() Pages { return onePage(col.Page()) }
+
+func (col *doubleColumnBuffer) Page() BufferedPage { return &col.doublePage }
+
+func (col *doubleColumnBuffer) Reset() { col.values = col.values[:0] }
+
+func (col *doubleColumnBuffer) Cap() int { return cap(col.values) }
+
+func (col *doubleColumnBuffer) Len() int { return len(col.values) }
+
+func (col *doubleColumnBuffer) Less(i, j int) bool { return col.values[i] < col.values[j] }
+
+func (col *doubleColumnBuffer) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *doubleColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 8) != 0 {
+		return 0, fmt.Errorf("cannot write DOUBLE values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToFloat64(b)...)
+	return len(b), nil
+}
+
+func (col *doubleColumnBuffer) WriteDoubles(values []float64) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
+	for _, v := range values {
+		col.values = append(col.values, v.Double())
+	}
+	return len(values), nil
+}
+
+func (col *doubleColumnBuffer) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, row[0].Double())
+	return nil
+}
+
+func (col *doubleColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := makeValueDouble(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
+
+type uint32ColumnBuffer struct{ *int32ColumnBuffer }
+
+func newUint32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint32ColumnBuffer {
+	return uint32ColumnBuffer{newInt32ColumnBuffer(typ, columnIndex, bufferSize)}
+}
+
+func (col uint32ColumnBuffer) ColumnIndex() ColumnIndex { return uint32ColumnIndex{col.page()} }
+
+func (col uint32ColumnBuffer) OffsetIndex() OffsetIndex { return uint32OffsetIndex{col.page()} }
+
+func (col uint32ColumnBuffer) page() uint32Page { return uint32Page{&col.int32Page} }
+
+func (col uint32ColumnBuffer) Page() BufferedPage { return col.page() }
+
+func (col uint32ColumnBuffer) Pages() Pages { return onePage(col.page()) }
+
+func (col uint32ColumnBuffer) Clone() ColumnBuffer {
+	return uint32ColumnBuffer{col.int32ColumnBuffer.Clone().(*int32ColumnBuffer)}
+}
+
+func (col uint32ColumnBuffer) Less(i, j int) bool {
+	return uint32(col.values[i]) < uint32(col.values[j])
+}
+
+type uint64ColumnBuffer struct{ *int64ColumnBuffer }
+
+func newUint64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint64ColumnBuffer {
+	return uint64ColumnBuffer{newInt64ColumnBuffer(typ, columnIndex, bufferSize)}
+}
+
+func (col uint64ColumnBuffer) ColumnIndex() ColumnIndex { return uint64ColumnIndex{col.page()} }
+
+func (col uint64ColumnBuffer) OffsetIndex() OffsetIndex { return uint64OffsetIndex{col.page()} }
+
+func (col uint64ColumnBuffer) page() uint64Page { return uint64Page{&col.int64Page} }
+
+func (col uint64ColumnBuffer) Page() BufferedPage { return col.page() }
+
+func (col uint64ColumnBuffer) Pages() Pages { return onePage(col.page()) }
+
+func (col uint64ColumnBuffer) Clone() ColumnBuffer {
+	return uint64ColumnBuffer{col.int64ColumnBuffer.Clone().(*int64ColumnBuffer)}
+}
+
+func (col uint64ColumnBuffer) Less(i, j int) bool {
+	return uint64(col.values[i]) < uint64(col.values[j])
+}
+
+var (
+	_ io.Writer = (*booleanColumnBuffer)(nil)
+	_ io.Writer = (*int32ColumnBuffer)(nil)
+	_ io.Writer = (*int64ColumnBuffer)(nil)
+	_ io.Writer = (*int96ColumnBuffer)(nil)
+	_ io.Writer = (*floatColumnBuffer)(nil)
+	_ io.Writer = (*doubleColumnBuffer)(nil)
+
+	_ BooleanWriter           = (*booleanColumnBuffer)(nil)
+	_ Int32Writer             = (*int32ColumnBuffer)(nil)
+	_ Int64Writer             = (*int64ColumnBuffer)(nil)
+	_ Int96Writer             = (*int96ColumnBuffer)(nil)
+	_ FloatWriter             = (*floatColumnBuffer)(nil)
+	_ DoubleWriter            = (*doubleColumnBuffer)(nil)
+	_ ByteArrayWriter         = (*byteArrayColumnBuffer)(nil)
+	_ FixedLenByteArrayWriter = (*fixedLenByteArrayColumnBuffer)(nil)
+)
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
new file mode 100644
index 0000000..86f3a72
--- /dev/null
+++ b/column_buffer_go18.go
@@ -0,0 +1,142 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+type columnBuffer[T primitive] struct {
+	page[T]
+	typ Type
+}
+
+func newBooleanColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[bool] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &boolClass)
+}
+
+func newInt32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[int32] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &int32Class)
+}
+
+func newInt64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[int64] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &int64Class)
+}
+
+func newInt96ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[deprecated.Int96] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &int96Class)
+}
+
+func newFloatColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[float32] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &float32Class)
+}
+
+func newDoubleColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[float64] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &float64Class)
+}
+
+func newUint32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[uint32] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &uint32Class)
+}
+
+func newUint64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) *columnBuffer[uint64] {
+	return newColumnBuffer(typ, columnIndex, bufferSize, &uint64Class)
+}
+
+func newColumnBuffer[T primitive](typ Type, columnIndex int16, bufferSize int, class *class[T]) *columnBuffer[T] {
+	return &columnBuffer[T]{
+		page: page[T]{
+			class:       class,
+			values:      make([]T, 0, bufferSize/sizeof[T]()),
+			columnIndex: ^columnIndex,
+		},
+		typ: typ,
+	}
+}
+
+func (col *columnBuffer[T]) Clone() ColumnBuffer {
+	return &columnBuffer[T]{
+		page: page[T]{
+			class:       col.page.class,
+			values:      append([]T{}, col.values...),
+			columnIndex: col.columnIndex,
+		},
+		typ: col.typ,
+	}
+}
+
+func (col *columnBuffer[T]) Type() Type { return col.typ }
+
+func (col *columnBuffer[T]) ColumnIndex() ColumnIndex { return columnIndex[T]{&col.page} }
+
+func (col *columnBuffer[T]) OffsetIndex() OffsetIndex { return offsetIndex[T]{&col.page} }
+
+func (col *columnBuffer[T]) BloomFilter() BloomFilter { return nil }
+
+func (col *columnBuffer[T]) Dictionary() Dictionary { return nil }
+
+func (col *columnBuffer[T]) Pages() Pages { return onePage(col.Page()) }
+
+func (col *columnBuffer[T]) Page() BufferedPage { return &col.page }
+
+func (col *columnBuffer[T]) Reset() { col.values = col.values[:0] }
+
+func (col *columnBuffer[T]) Cap() int { return cap(col.values) }
+
+func (col *columnBuffer[T]) Len() int { return len(col.values) }
+
+func (col *columnBuffer[T]) Less(i, j int) bool {
+	return col.class.less(col.values[i], col.values[j])
+}
+
+func (col *columnBuffer[T]) Swap(i, j int) {
+	col.values[i], col.values[j] = col.values[j], col.values[i]
+}
+
+func (col *columnBuffer[T]) Write(b []byte) (int, error) {
+	if (len(b) % sizeof[T]()) != 0 {
+		return 0, fmt.Errorf("cannot write %s values from input of size %d", col.class.name, len(b))
+	}
+	return col.WriteRequired(cast.BytesToSlice[T](b))
+}
+
+func (col *columnBuffer[T]) WriteRequired(values []T) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
+func (col *columnBuffer[T]) WriteValues(values []Value) (int, error) {
+	value := col.class.value
+	for _, v := range values {
+		col.values = append(col.values, value(v))
+	}
+	return len(values), nil
+}
+
+func (col *columnBuffer[T]) WriteRow(row Row) error {
+	if len(row) == 0 {
+		return errRowHasTooFewValues(int64(len(row)))
+	}
+	if len(row) > 1 {
+		return errRowHasTooManyValues(int64(len(row)))
+	}
+	col.values = append(col.values, col.class.value(row[0]))
+	return nil
+}
+
+func (col *columnBuffer[T]) ReadRowAt(row Row, index int64) (Row, error) {
+	switch {
+	case index < 0:
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
+		return row, io.EOF
+	default:
+		v := col.class.makeValue(col.values[index])
+		v.columnIndex = col.columnIndex
+		return append(row, v), nil
+	}
+}
diff --git a/column_index.go b/column_index.go
index 6a1b2d7..d23387b 100644
--- a/column_index.go
+++ b/column_index.go
@@ -3,9 +3,7 @@ package parquet
 import (
 	"bytes"
 
-	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/encoding/plain"
 	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
@@ -22,8 +20,8 @@ type ColumnIndex interface {
 
 	// PageIndex return min/max bounds for the page at the given index in the
 	// column.
-	MinValue(int) []byte
-	MaxValue(int) []byte
+	MinValue(int) Value
+	MaxValue(int) Value
 
 	// IsAscending returns true if the column index min/max values are sorted
 	// in ascending order (based on the ordering rules of the column's logical
@@ -36,119 +34,56 @@ type ColumnIndex interface {
 	IsDescending() bool
 }
 
-type columnIndex format.ColumnIndex
-
-func (index *columnIndex) NumPages() int         { return len(index.NullPages) }
-func (index *columnIndex) NullCount(i int) int64 { return index.NullCounts[i] }
-func (index *columnIndex) NullPage(i int) bool   { return index.NullPages[i] }
-func (index *columnIndex) MinValue(i int) []byte { return index.MinValues[i] }
-func (index *columnIndex) MaxValue(i int) []byte { return index.MaxValues[i] }
-func (index *columnIndex) IsAscending() bool     { return index.BoundaryOrder == format.Ascending }
-func (index *columnIndex) IsDescending() bool    { return index.BoundaryOrder == format.Descending }
-
-type booleanPageIndex struct{ page *booleanPage }
-
-func (index booleanPageIndex) NumPages() int       { return 1 }
-func (index booleanPageIndex) NullCount(int) int64 { return 0 }
-func (index booleanPageIndex) NullPage(int) bool   { return false }
-func (index booleanPageIndex) MinValue(int) []byte { return plain.Boolean(index.page.min()) }
-func (index booleanPageIndex) MaxValue(int) []byte { return plain.Boolean(index.page.max()) }
-func (index booleanPageIndex) IsAscending() bool   { return compareBool(index.page.bounds()) < 0 }
-func (index booleanPageIndex) IsDescending() bool  { return compareBool(index.page.bounds()) > 0 }
-
-type int32PageIndex struct{ page *int32Page }
-
-func (index int32PageIndex) NumPages() int       { return 1 }
-func (index int32PageIndex) NullCount(int) int64 { return 0 }
-func (index int32PageIndex) NullPage(int) bool   { return false }
-func (index int32PageIndex) MinValue(int) []byte { return plain.Int32(index.page.min()) }
-func (index int32PageIndex) MaxValue(int) []byte { return plain.Int32(index.page.max()) }
-func (index int32PageIndex) IsAscending() bool   { return compareInt32(index.page.bounds()) < 0 }
-func (index int32PageIndex) IsDescending() bool  { return compareInt32(index.page.bounds()) > 0 }
-
-type int64PageIndex struct{ page *int64Page }
-
-func (index int64PageIndex) NumPages() int       { return 1 }
-func (index int64PageIndex) NullCount(int) int64 { return 0 }
-func (index int64PageIndex) NullPage(int) bool   { return false }
-func (index int64PageIndex) MinValue(int) []byte { return plain.Int64(index.page.min()) }
-func (index int64PageIndex) MaxValue(int) []byte { return plain.Int64(index.page.max()) }
-func (index int64PageIndex) IsAscending() bool   { return compareInt64(index.page.bounds()) < 0 }
-func (index int64PageIndex) IsDescending() bool  { return compareInt64(index.page.bounds()) > 0 }
-
-type int96PageIndex struct{ page *int96Page }
-
-func (index int96PageIndex) NumPages() int       { return 1 }
-func (index int96PageIndex) NullCount(int) int64 { return 0 }
-func (index int96PageIndex) NullPage(int) bool   { return false }
-func (index int96PageIndex) MinValue(int) []byte { return plain.Int96(index.page.min()) }
-func (index int96PageIndex) MaxValue(int) []byte { return plain.Int96(index.page.max()) }
-func (index int96PageIndex) IsAscending() bool   { return compareInt96(index.page.bounds()) < 0 }
-func (index int96PageIndex) IsDescending() bool  { return compareInt96(index.page.bounds()) > 0 }
-
-type floatPageIndex struct{ page *floatPage }
-
-func (index floatPageIndex) NumPages() int       { return 1 }
-func (index floatPageIndex) NullCount(int) int64 { return 0 }
-func (index floatPageIndex) NullPage(int) bool   { return false }
-func (index floatPageIndex) MinValue(int) []byte { return plain.Float(index.page.min()) }
-func (index floatPageIndex) MaxValue(int) []byte { return plain.Float(index.page.max()) }
-func (index floatPageIndex) IsAscending() bool   { return compareFloat32(index.page.bounds()) < 0 }
-func (index floatPageIndex) IsDescending() bool  { return compareFloat32(index.page.bounds()) > 0 }
-
-type doublePageIndex struct{ page *doublePage }
-
-func (index doublePageIndex) NumPages() int       { return 1 }
-func (index doublePageIndex) NullCount(int) int64 { return 0 }
-func (index doublePageIndex) NullPage(int) bool   { return false }
-func (index doublePageIndex) MinValue(int) []byte { return plain.Double(index.page.min()) }
-func (index doublePageIndex) MaxValue(int) []byte { return plain.Double(index.page.max()) }
-func (index doublePageIndex) IsAscending() bool   { return compareFloat64(index.page.bounds()) < 0 }
-func (index doublePageIndex) IsDescending() bool  { return compareFloat64(index.page.bounds()) > 0 }
-
-type byteArrayPageIndex struct{ page *byteArrayPage }
-
-func (index byteArrayPageIndex) NumPages() int       { return 1 }
-func (index byteArrayPageIndex) NullCount(int) int64 { return 0 }
-func (index byteArrayPageIndex) NullPage(int) bool   { return false }
-func (index byteArrayPageIndex) MinValue(int) []byte { return copyBytes(index.page.min()) }
-func (index byteArrayPageIndex) MaxValue(int) []byte { return copyBytes(index.page.max()) }
-func (index byteArrayPageIndex) IsAscending() bool   { return bytes.Compare(index.page.bounds()) < 0 }
-func (index byteArrayPageIndex) IsDescending() bool  { return bytes.Compare(index.page.bounds()) > 0 }
-
-type fixedLenByteArrayPageIndex struct{ page *fixedLenByteArrayPage }
-
-func (index fixedLenByteArrayPageIndex) NumPages() int       { return 1 }
-func (index fixedLenByteArrayPageIndex) NullCount(int) int64 { return 0 }
-func (index fixedLenByteArrayPageIndex) NullPage(int) bool   { return false }
-func (index fixedLenByteArrayPageIndex) MinValue(int) []byte { return copyBytes(index.page.min()) }
-func (index fixedLenByteArrayPageIndex) MaxValue(int) []byte { return copyBytes(index.page.max()) }
-func (index fixedLenByteArrayPageIndex) IsAscending() bool {
-	return bytes.Compare(index.page.bounds()) < 0
+type emptyColumnIndex struct{}
+
+func (emptyColumnIndex) NumPages() int       { return 0 }
+func (emptyColumnIndex) NullCount(int) int64 { return 0 }
+func (emptyColumnIndex) NullPage(int) bool   { return false }
+func (emptyColumnIndex) MinValue(int) Value  { return Value{} }
+func (emptyColumnIndex) MaxValue(int) Value  { return Value{} }
+func (emptyColumnIndex) IsAscending() bool   { return false }
+func (emptyColumnIndex) IsDescending() bool  { return false }
+
+type fileColumnIndex struct{ chunk *fileColumnChunk }
+
+func (i fileColumnIndex) NumPages() int         { return len(i.chunk.columnIndex.NullPages) }
+func (i fileColumnIndex) NullCount(j int) int64 { return i.chunk.columnIndex.NullCounts[j] }
+func (i fileColumnIndex) NullPage(j int) bool   { return i.chunk.columnIndex.NullPages[j] }
+func (i fileColumnIndex) MinValue(j int) Value  { return i.makeValue(i.chunk.columnIndex.MinValues[j]) }
+func (i fileColumnIndex) MaxValue(j int) Value  { return i.makeValue(i.chunk.columnIndex.MaxValues[j]) }
+func (i fileColumnIndex) IsAscending() bool {
+	return i.chunk.columnIndex.BoundaryOrder == format.Ascending
 }
-func (index fixedLenByteArrayPageIndex) IsDescending() bool {
-	return bytes.Compare(index.page.bounds()) > 0
+func (i fileColumnIndex) IsDescending() bool {
+	return i.chunk.columnIndex.BoundaryOrder == format.Descending
+}
+func (i *fileColumnIndex) makeValue(b []byte) Value {
+	return i.chunk.column.typ.Kind().Value(b)
 }
 
-type uint32PageIndex struct{ page uint32Page }
+type byteArrayColumnIndex struct{ page *byteArrayPage }
 
-func (index uint32PageIndex) NumPages() int       { return 1 }
-func (index uint32PageIndex) NullCount(int) int64 { return 0 }
-func (index uint32PageIndex) NullPage(int) bool   { return false }
-func (index uint32PageIndex) MinValue(int) []byte { return plain.Int32(int32(index.page.min())) }
-func (index uint32PageIndex) MaxValue(int) []byte { return plain.Int32(int32(index.page.max())) }
-func (index uint32PageIndex) IsAscending() bool   { return compareUint32(index.page.bounds()) < 0 }
-func (index uint32PageIndex) IsDescending() bool  { return compareUint32(index.page.bounds()) > 0 }
+func (i byteArrayColumnIndex) NumPages() int       { return 1 }
+func (i byteArrayColumnIndex) NullCount(int) int64 { return 0 }
+func (i byteArrayColumnIndex) NullPage(int) bool   { return false }
+func (i byteArrayColumnIndex) MinValue(int) Value  { return makeValueBytes(ByteArray, i.page.min()) }
+func (i byteArrayColumnIndex) MaxValue(int) Value  { return makeValueBytes(ByteArray, i.page.max()) }
+func (i byteArrayColumnIndex) IsAscending() bool   { return bytes.Compare(i.page.bounds()) < 0 }
+func (i byteArrayColumnIndex) IsDescending() bool  { return bytes.Compare(i.page.bounds()) > 0 }
 
-type uint64PageIndex struct{ page uint64Page }
+type fixedLenByteArrayColumnIndex struct{ page *fixedLenByteArrayPage }
 
-func (index uint64PageIndex) NumPages() int       { return 1 }
-func (index uint64PageIndex) NullCount(int) int64 { return 0 }
-func (index uint64PageIndex) NullPage(int) bool   { return false }
-func (index uint64PageIndex) MinValue(int) []byte { return plain.Int64(int64(index.page.min())) }
-func (index uint64PageIndex) MaxValue(int) []byte { return plain.Int64(int64(index.page.max())) }
-func (index uint64PageIndex) IsAscending() bool   { return compareUint64(index.page.bounds()) < 0 }
-func (index uint64PageIndex) IsDescending() bool  { return compareUint64(index.page.bounds()) > 0 }
+func (i fixedLenByteArrayColumnIndex) NumPages() int       { return 1 }
+func (i fixedLenByteArrayColumnIndex) NullCount(int) int64 { return 0 }
+func (i fixedLenByteArrayColumnIndex) NullPage(int) bool   { return false }
+func (i fixedLenByteArrayColumnIndex) MinValue(int) Value {
+	return makeValueBytes(FixedLenByteArray, i.page.min())
+}
+func (i fixedLenByteArrayColumnIndex) MaxValue(int) Value {
+	return makeValueBytes(FixedLenByteArray, i.page.max())
+}
+func (i fixedLenByteArrayColumnIndex) IsAscending() bool  { return bytes.Compare(i.page.bounds()) < 0 }
+func (i fixedLenByteArrayColumnIndex) IsDescending() bool { return bytes.Compare(i.page.bounds()) > 0 }
 
 // The ColumnIndexer interface is implemented by types that support generating
 // parquet column indexes.
@@ -171,219 +106,33 @@ type ColumnIndexer interface {
 	ColumnIndex() format.ColumnIndex
 }
 
-type columnIndexer struct {
+type baseColumnIndexer struct {
 	nullPages  []bool
 	nullCounts []int64
 }
 
-func (index *columnIndexer) reset() {
-	index.nullPages = index.nullPages[:0]
-	index.nullCounts = index.nullCounts[:0]
+func (i *baseColumnIndexer) reset() {
+	i.nullPages = i.nullPages[:0]
+	i.nullCounts = i.nullCounts[:0]
 }
 
-func (index *columnIndexer) observe(numValues, numNulls int64) {
-	index.nullPages = append(index.nullPages, numValues == numNulls)
-	index.nullCounts = append(index.nullCounts, numNulls)
+func (i *baseColumnIndexer) observe(numValues, numNulls int64) {
+	i.nullPages = append(i.nullPages, numValues == numNulls)
+	i.nullCounts = append(i.nullCounts, numNulls)
 }
 
-func (index *columnIndexer) columnIndex(minValues, maxValues [][]byte, minOrder, maxOrder int) format.ColumnIndex {
+func (i *baseColumnIndexer) columnIndex(minValues, maxValues [][]byte, minOrder, maxOrder int) format.ColumnIndex {
 	return format.ColumnIndex{
-		NullPages:     index.nullPages,
-		NullCounts:    index.nullCounts,
+		NullPages:     i.nullPages,
+		NullCounts:    i.nullCounts,
 		MinValues:     minValues,
 		MaxValues:     maxValues,
 		BoundaryOrder: boundaryOrderOf(minOrder, maxOrder),
 	}
 }
 
-type booleanColumnIndexer struct {
-	columnIndexer
-	minValues []bool
-	maxValues []bool
-}
-
-func newBooleanColumnIndexer() *booleanColumnIndexer {
-	return new(booleanColumnIndexer)
-}
-
-func (index *booleanColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *booleanColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Boolean())
-	index.maxValues = append(index.maxValues, max.Boolean())
-}
-
-func (index *booleanColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(1, bits.BoolToBytes(index.minValues)),
-		splitFixedLenByteArrayList(1, bits.BoolToBytes(index.maxValues)),
-		bits.OrderOfBool(index.minValues),
-		bits.OrderOfBool(index.maxValues),
-	)
-}
-
-type int32ColumnIndexer struct {
-	columnIndexer
-	minValues []int32
-	maxValues []int32
-}
-
-func newInt32ColumnIndexer() *int32ColumnIndexer {
-	return new(int32ColumnIndexer)
-}
-
-func (index *int32ColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *int32ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Int32())
-	index.maxValues = append(index.maxValues, max.Int32())
-}
-
-func (index *int32ColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(4, bits.Int32ToBytes(index.minValues)),
-		splitFixedLenByteArrayList(4, bits.Int32ToBytes(index.maxValues)),
-		bits.OrderOfInt32(index.minValues),
-		bits.OrderOfInt32(index.maxValues),
-	)
-}
-
-type int64ColumnIndexer struct {
-	columnIndexer
-	minValues []int64
-	maxValues []int64
-}
-
-func newInt64ColumnIndexer() *int64ColumnIndexer {
-	return new(int64ColumnIndexer)
-}
-
-func (index *int64ColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *int64ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Int64())
-	index.maxValues = append(index.maxValues, max.Int64())
-}
-
-func (index *int64ColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(8, bits.Int64ToBytes(index.minValues)),
-		splitFixedLenByteArrayList(8, bits.Int64ToBytes(index.maxValues)),
-		bits.OrderOfInt64(index.minValues),
-		bits.OrderOfInt64(index.maxValues),
-	)
-}
-
-type int96ColumnIndexer struct {
-	columnIndexer
-	minValues []deprecated.Int96
-	maxValues []deprecated.Int96
-}
-
-func newInt96ColumnIndexer() *int96ColumnIndexer {
-	return new(int96ColumnIndexer)
-}
-
-func (index *int96ColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *int96ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Int96())
-	index.maxValues = append(index.maxValues, max.Int96())
-}
-
-func (index *int96ColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(12, deprecated.Int96ToBytes(index.minValues)),
-		splitFixedLenByteArrayList(12, deprecated.Int96ToBytes(index.maxValues)),
-		deprecated.OrderOfInt96(index.minValues),
-		deprecated.OrderOfInt96(index.maxValues),
-	)
-}
-
-type floatColumnIndexer struct {
-	columnIndexer
-	minValues []float32
-	maxValues []float32
-}
-
-func newFloatColumnIndexer() *floatColumnIndexer {
-	return new(floatColumnIndexer)
-}
-
-func (index *floatColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *floatColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Float())
-	index.maxValues = append(index.maxValues, max.Float())
-}
-
-func (index *floatColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(4, bits.Float32ToBytes(index.minValues)),
-		splitFixedLenByteArrayList(4, bits.Float32ToBytes(index.maxValues)),
-		bits.OrderOfFloat32(index.minValues),
-		bits.OrderOfFloat32(index.maxValues),
-	)
-}
-
-type doubleColumnIndexer struct {
-	columnIndexer
-	minValues []float64
-	maxValues []float64
-}
-
-func newDoubleColumnIndexer() *doubleColumnIndexer {
-	return new(doubleColumnIndexer)
-}
-
-func (index *doubleColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
-}
-
-func (index *doubleColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.Double())
-	index.maxValues = append(index.maxValues, max.Double())
-}
-
-func (index *doubleColumnIndexer) ColumnIndex() format.ColumnIndex {
-	return index.columnIndex(
-		splitFixedLenByteArrayList(8, bits.Float64ToBytes(index.minValues)),
-		splitFixedLenByteArrayList(8, bits.Float64ToBytes(index.maxValues)),
-		bits.OrderOfFloat64(index.minValues),
-		bits.OrderOfFloat64(index.maxValues),
-	)
-}
-
 type byteArrayColumnIndexer struct {
-	columnIndexer
+	baseColumnIndexer
 	sizeLimit int
 	minValues encoding.ByteArrayList
 	maxValues encoding.ByteArrayList
@@ -393,26 +142,26 @@ func newByteArrayColumnIndexer(sizeLimit int) *byteArrayColumnIndexer {
 	return &byteArrayColumnIndexer{sizeLimit: sizeLimit}
 }
 
-func (index *byteArrayColumnIndexer) Reset() {
-	index.reset()
-	index.minValues.Reset()
-	index.maxValues.Reset()
+func (i *byteArrayColumnIndexer) Reset() {
+	i.reset()
+	i.minValues.Reset()
+	i.maxValues.Reset()
 }
 
-func (index *byteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues.Push(min.ByteArray())
-	index.maxValues.Push(max.ByteArray())
+func (i *byteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues.Push(min.ByteArray())
+	i.maxValues.Push(max.ByteArray())
 }
 
-func (index *byteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
-	minValues := index.minValues.Split()
-	maxValues := index.maxValues.Split()
-	if index.sizeLimit > 0 {
-		truncateLargeMinByteArrayValues(minValues, index.sizeLimit)
-		truncateLargeMaxByteArrayValues(maxValues, index.sizeLimit)
+func (i *byteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
+	minValues := i.minValues.Split()
+	maxValues := i.maxValues.Split()
+	if i.sizeLimit > 0 {
+		truncateLargeMinByteArrayValues(minValues, i.sizeLimit)
+		truncateLargeMaxByteArrayValues(maxValues, i.sizeLimit)
 	}
-	return index.columnIndex(
+	return i.columnIndex(
 		minValues,
 		maxValues,
 		bits.OrderOfBytes(minValues),
@@ -485,7 +234,7 @@ func nextByteArrayValue(value []byte) []byte {
 }
 
 type fixedLenByteArrayColumnIndexer struct {
-	columnIndexer
+	baseColumnIndexer
 	size      int
 	sizeLimit int
 	minValues []byte
@@ -499,26 +248,26 @@ func newFixedLenByteArrayColumnIndexer(size, sizeLimit int) *fixedLenByteArrayCo
 	}
 }
 
-func (index *fixedLenByteArrayColumnIndexer) Reset() {
-	index.reset()
-	index.minValues = index.minValues[:0]
-	index.maxValues = index.maxValues[:0]
+func (i *fixedLenByteArrayColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
 }
 
-func (index *fixedLenByteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
-	index.observe(numValues, numNulls)
-	index.minValues = append(index.minValues, min.ByteArray()...)
-	index.maxValues = append(index.maxValues, max.ByteArray()...)
+func (i *fixedLenByteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.ByteArray()...)
+	i.maxValues = append(i.maxValues, max.ByteArray()...)
 }
 
-func (index *fixedLenByteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
-	minValues := splitFixedLenByteArrayList(index.size, index.minValues)
-	maxValues := splitFixedLenByteArrayList(index.size, index.maxValues)
-	if index.sizeLimit > 0 && index.sizeLimit < index.size {
-		truncateLargeMinByteArrayValues(minValues, index.sizeLimit)
-		truncateLargeMaxByteArrayValues(maxValues, index.sizeLimit)
+func (i *fixedLenByteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
+	minValues := splitFixedLenByteArrayList(i.size, i.minValues)
+	maxValues := splitFixedLenByteArrayList(i.size, i.maxValues)
+	if i.sizeLimit > 0 && i.sizeLimit < i.size {
+		truncateLargeMinByteArrayValues(minValues, i.sizeLimit)
+		truncateLargeMaxByteArrayValues(maxValues, i.sizeLimit)
 	}
-	return index.columnIndex(
+	return i.columnIndex(
 		minValues,
 		maxValues,
 		bits.OrderOfBytes(minValues),
@@ -526,40 +275,6 @@ func (index *fixedLenByteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
 	)
 }
 
-type uint32ColumnIndexer struct{ *int32ColumnIndexer }
-
-func newUint32ColumnIndexer() uint32ColumnIndexer {
-	return uint32ColumnIndexer{newInt32ColumnIndexer()}
-}
-
-func (index uint32ColumnIndexer) ColumnIndex() format.ColumnIndex {
-	minValues := bits.Int32ToUint32(index.minValues)
-	maxValues := bits.Int32ToUint32(index.maxValues)
-	return index.columnIndex(
-		splitFixedLenByteArrayList(4, bits.Uint32ToBytes(minValues)),
-		splitFixedLenByteArrayList(4, bits.Uint32ToBytes(maxValues)),
-		bits.OrderOfUint32(minValues),
-		bits.OrderOfUint32(maxValues),
-	)
-}
-
-type uint64ColumnIndexer struct{ *int64ColumnIndexer }
-
-func newUint64ColumnIndexer() uint64ColumnIndexer {
-	return uint64ColumnIndexer{newInt64ColumnIndexer()}
-}
-
-func (index uint64ColumnIndexer) ColumnIndex() format.ColumnIndex {
-	minValues := bits.Int64ToUint64(index.minValues)
-	maxValues := bits.Int64ToUint64(index.maxValues)
-	return index.columnIndex(
-		splitFixedLenByteArrayList(8, bits.Uint64ToBytes(minValues)),
-		splitFixedLenByteArrayList(8, bits.Uint64ToBytes(maxValues)),
-		bits.OrderOfUint64(minValues),
-		bits.OrderOfUint64(maxValues),
-	)
-}
-
 func splitFixedLenByteArrayList(size int, data []byte) [][]byte {
 	data = copyBytes(data)
 	values := make([][]byte, len(data)/size)
diff --git a/column_index_default.go b/column_index_default.go
new file mode 100644
index 0000000..d9bace2
--- /dev/null
+++ b/column_index_default.go
@@ -0,0 +1,309 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/format"
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+type booleanColumnIndex struct{ page *booleanPage }
+
+func (i booleanColumnIndex) NumPages() int       { return 1 }
+func (i booleanColumnIndex) NullCount(int) int64 { return 0 }
+func (i booleanColumnIndex) NullPage(int) bool   { return false }
+func (i booleanColumnIndex) MinValue(int) Value  { return makeValueBoolean(i.page.min()) }
+func (i booleanColumnIndex) MaxValue(int) Value  { return makeValueBoolean(i.page.max()) }
+func (i booleanColumnIndex) IsAscending() bool   { return compareBool(i.page.bounds()) < 0 }
+func (i booleanColumnIndex) IsDescending() bool  { return compareBool(i.page.bounds()) > 0 }
+
+type int32ColumnIndex struct{ page *int32Page }
+
+func (i int32ColumnIndex) NumPages() int       { return 1 }
+func (i int32ColumnIndex) NullCount(int) int64 { return 0 }
+func (i int32ColumnIndex) NullPage(int) bool   { return false }
+func (i int32ColumnIndex) MinValue(int) Value  { return makeValueInt32(i.page.min()) }
+func (i int32ColumnIndex) MaxValue(int) Value  { return makeValueInt32(i.page.max()) }
+func (i int32ColumnIndex) IsAscending() bool   { return compareInt32(i.page.bounds()) < 0 }
+func (i int32ColumnIndex) IsDescending() bool  { return compareInt32(i.page.bounds()) > 0 }
+
+type int64ColumnIndex struct{ page *int64Page }
+
+func (i int64ColumnIndex) NumPages() int       { return 1 }
+func (i int64ColumnIndex) NullCount(int) int64 { return 0 }
+func (i int64ColumnIndex) NullPage(int) bool   { return false }
+func (i int64ColumnIndex) MinValue(int) Value  { return makeValueInt64(i.page.min()) }
+func (i int64ColumnIndex) MaxValue(int) Value  { return makeValueInt64(i.page.max()) }
+func (i int64ColumnIndex) IsAscending() bool   { return compareInt64(i.page.bounds()) < 0 }
+func (i int64ColumnIndex) IsDescending() bool  { return compareInt64(i.page.bounds()) > 0 }
+
+type int96ColumnIndex struct{ page *int96Page }
+
+func (i int96ColumnIndex) NumPages() int       { return 1 }
+func (i int96ColumnIndex) NullCount(int) int64 { return 0 }
+func (i int96ColumnIndex) NullPage(int) bool   { return false }
+func (i int96ColumnIndex) MinValue(int) Value  { return makeValueInt96(i.page.min()) }
+func (i int96ColumnIndex) MaxValue(int) Value  { return makeValueInt96(i.page.max()) }
+func (i int96ColumnIndex) IsAscending() bool   { return compareInt96(i.page.bounds()) < 0 }
+func (i int96ColumnIndex) IsDescending() bool  { return compareInt96(i.page.bounds()) > 0 }
+
+type floatColumnIndex struct{ page *floatPage }
+
+func (i floatColumnIndex) NumPages() int       { return 1 }
+func (i floatColumnIndex) NullCount(int) int64 { return 0 }
+func (i floatColumnIndex) NullPage(int) bool   { return false }
+func (i floatColumnIndex) MinValue(int) Value  { return makeValueFloat(i.page.min()) }
+func (i floatColumnIndex) MaxValue(int) Value  { return makeValueFloat(i.page.max()) }
+func (i floatColumnIndex) IsAscending() bool   { return compareFloat32(i.page.bounds()) < 0 }
+func (i floatColumnIndex) IsDescending() bool  { return compareFloat32(i.page.bounds()) > 0 }
+
+type doubleColumnIndex struct{ page *doublePage }
+
+func (i doubleColumnIndex) NumPages() int       { return 1 }
+func (i doubleColumnIndex) NullCount(int) int64 { return 0 }
+func (i doubleColumnIndex) NullPage(int) bool   { return false }
+func (i doubleColumnIndex) MinValue(int) Value  { return makeValueDouble(i.page.min()) }
+func (i doubleColumnIndex) MaxValue(int) Value  { return makeValueDouble(i.page.max()) }
+func (i doubleColumnIndex) IsAscending() bool   { return compareFloat64(i.page.bounds()) < 0 }
+func (i doubleColumnIndex) IsDescending() bool  { return compareFloat64(i.page.bounds()) > 0 }
+
+type uint32ColumnIndex struct{ page uint32Page }
+
+func (i uint32ColumnIndex) NumPages() int       { return 1 }
+func (i uint32ColumnIndex) NullCount(int) int64 { return 0 }
+func (i uint32ColumnIndex) NullPage(int) bool   { return false }
+func (i uint32ColumnIndex) MinValue(int) Value  { return makeValueInt32(int32(i.page.min())) }
+func (i uint32ColumnIndex) MaxValue(int) Value  { return makeValueInt32(int32(i.page.max())) }
+func (i uint32ColumnIndex) IsAscending() bool   { return compareUint32(i.page.bounds()) < 0 }
+func (i uint32ColumnIndex) IsDescending() bool  { return compareUint32(i.page.bounds()) > 0 }
+
+type uint64ColumnIndex struct{ page uint64Page }
+
+func (i uint64ColumnIndex) NumPages() int       { return 1 }
+func (i uint64ColumnIndex) NullCount(int) int64 { return 0 }
+func (i uint64ColumnIndex) NullPage(int) bool   { return false }
+func (i uint64ColumnIndex) MinValue(int) Value  { return makeValueInt64(int64(i.page.min())) }
+func (i uint64ColumnIndex) MaxValue(int) Value  { return makeValueInt64(int64(i.page.max())) }
+func (i uint64ColumnIndex) IsAscending() bool   { return compareUint64(i.page.bounds()) < 0 }
+func (i uint64ColumnIndex) IsDescending() bool  { return compareUint64(i.page.bounds()) > 0 }
+
+type booleanColumnIndexer struct {
+	baseColumnIndexer
+	minValues []bool
+	maxValues []bool
+}
+
+func newBooleanColumnIndexer() *booleanColumnIndexer {
+	return new(booleanColumnIndexer)
+}
+
+func (i *booleanColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *booleanColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Boolean())
+	i.maxValues = append(i.maxValues, max.Boolean())
+}
+
+func (i *booleanColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(1, bits.BoolToBytes(i.minValues)),
+		splitFixedLenByteArrayList(1, bits.BoolToBytes(i.maxValues)),
+		bits.OrderOfBool(i.minValues),
+		bits.OrderOfBool(i.maxValues),
+	)
+}
+
+type int32ColumnIndexer struct {
+	baseColumnIndexer
+	minValues []int32
+	maxValues []int32
+}
+
+func newInt32ColumnIndexer() *int32ColumnIndexer {
+	return new(int32ColumnIndexer)
+}
+
+func (i *int32ColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *int32ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Int32())
+	i.maxValues = append(i.maxValues, max.Int32())
+}
+
+func (i *int32ColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(4, bits.Int32ToBytes(i.minValues)),
+		splitFixedLenByteArrayList(4, bits.Int32ToBytes(i.maxValues)),
+		bits.OrderOfInt32(i.minValues),
+		bits.OrderOfInt32(i.maxValues),
+	)
+}
+
+type int64ColumnIndexer struct {
+	baseColumnIndexer
+	minValues []int64
+	maxValues []int64
+}
+
+func newInt64ColumnIndexer() *int64ColumnIndexer {
+	return new(int64ColumnIndexer)
+}
+
+func (i *int64ColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *int64ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Int64())
+	i.maxValues = append(i.maxValues, max.Int64())
+}
+
+func (i *int64ColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(8, bits.Int64ToBytes(i.minValues)),
+		splitFixedLenByteArrayList(8, bits.Int64ToBytes(i.maxValues)),
+		bits.OrderOfInt64(i.minValues),
+		bits.OrderOfInt64(i.maxValues),
+	)
+}
+
+type int96ColumnIndexer struct {
+	baseColumnIndexer
+	minValues []deprecated.Int96
+	maxValues []deprecated.Int96
+}
+
+func newInt96ColumnIndexer() *int96ColumnIndexer {
+	return new(int96ColumnIndexer)
+}
+
+func (i *int96ColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *int96ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Int96())
+	i.maxValues = append(i.maxValues, max.Int96())
+}
+
+func (i *int96ColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(12, deprecated.Int96ToBytes(i.minValues)),
+		splitFixedLenByteArrayList(12, deprecated.Int96ToBytes(i.maxValues)),
+		deprecated.OrderOfInt96(i.minValues),
+		deprecated.OrderOfInt96(i.maxValues),
+	)
+}
+
+type floatColumnIndexer struct {
+	baseColumnIndexer
+	minValues []float32
+	maxValues []float32
+}
+
+func newFloatColumnIndexer() *floatColumnIndexer {
+	return new(floatColumnIndexer)
+}
+
+func (i *floatColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *floatColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Float())
+	i.maxValues = append(i.maxValues, max.Float())
+}
+
+func (i *floatColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(4, bits.Float32ToBytes(i.minValues)),
+		splitFixedLenByteArrayList(4, bits.Float32ToBytes(i.maxValues)),
+		bits.OrderOfFloat32(i.minValues),
+		bits.OrderOfFloat32(i.maxValues),
+	)
+}
+
+type doubleColumnIndexer struct {
+	baseColumnIndexer
+	minValues []float64
+	maxValues []float64
+}
+
+func newDoubleColumnIndexer() *doubleColumnIndexer {
+	return new(doubleColumnIndexer)
+}
+
+func (i *doubleColumnIndexer) Reset() {
+	i.reset()
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *doubleColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.observe(numValues, numNulls)
+	i.minValues = append(i.minValues, min.Double())
+	i.maxValues = append(i.maxValues, max.Double())
+}
+
+func (i *doubleColumnIndexer) ColumnIndex() format.ColumnIndex {
+	return i.columnIndex(
+		splitFixedLenByteArrayList(8, bits.Float64ToBytes(i.minValues)),
+		splitFixedLenByteArrayList(8, bits.Float64ToBytes(i.maxValues)),
+		bits.OrderOfFloat64(i.minValues),
+		bits.OrderOfFloat64(i.maxValues),
+	)
+}
+
+type uint32ColumnIndexer struct{ *int32ColumnIndexer }
+
+func newUint32ColumnIndexer() uint32ColumnIndexer {
+	return uint32ColumnIndexer{newInt32ColumnIndexer()}
+}
+
+func (i uint32ColumnIndexer) ColumnIndex() format.ColumnIndex {
+	minValues := bits.Int32ToUint32(i.minValues)
+	maxValues := bits.Int32ToUint32(i.maxValues)
+	return i.columnIndex(
+		splitFixedLenByteArrayList(4, bits.Uint32ToBytes(minValues)),
+		splitFixedLenByteArrayList(4, bits.Uint32ToBytes(maxValues)),
+		bits.OrderOfUint32(minValues),
+		bits.OrderOfUint32(maxValues),
+	)
+}
+
+type uint64ColumnIndexer struct{ *int64ColumnIndexer }
+
+func newUint64ColumnIndexer() uint64ColumnIndexer {
+	return uint64ColumnIndexer{newInt64ColumnIndexer()}
+}
+
+func (i uint64ColumnIndexer) ColumnIndex() format.ColumnIndex {
+	minValues := bits.Int64ToUint64(i.minValues)
+	maxValues := bits.Int64ToUint64(i.maxValues)
+	return i.columnIndex(
+		splitFixedLenByteArrayList(8, bits.Uint64ToBytes(minValues)),
+		splitFixedLenByteArrayList(8, bits.Uint64ToBytes(maxValues)),
+		bits.OrderOfUint64(minValues),
+		bits.OrderOfUint64(maxValues),
+	)
+}
diff --git a/column_index_go18.go b/column_index_go18.go
new file mode 100644
index 0000000..95cb2c3
--- /dev/null
+++ b/column_index_go18.go
@@ -0,0 +1,87 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/format"
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+type columnIndex[T primitive] struct{ page *page[T] }
+
+func (i columnIndex[T]) NumPages() int       { return 1 }
+func (i columnIndex[T]) NullCount(int) int64 { return 0 }
+func (i columnIndex[T]) NullPage(int) bool   { return false }
+func (i columnIndex[T]) MinValue(int) Value  { return i.page.class.makeValue(i.page.min()) }
+func (i columnIndex[T]) MaxValue(int) Value  { return i.page.class.makeValue(i.page.max()) }
+func (i columnIndex[T]) IsAscending() bool   { return i.page.class.compare(i.page.bounds()) < 0 }
+func (i columnIndex[T]) IsDescending() bool  { return i.page.class.compare(i.page.bounds()) > 0 }
+
+type columnIndexer[T primitive] struct {
+	class      *class[T]
+	nullPages  []bool
+	nullCounts []int64
+	minValues  []T
+	maxValues  []T
+}
+
+func (i *columnIndexer[T]) Reset() {
+	i.nullPages = i.nullPages[:0]
+	i.nullCounts = i.nullCounts[:0]
+	i.minValues = i.minValues[:0]
+	i.maxValues = i.maxValues[:0]
+}
+
+func (i *columnIndexer[T]) IndexPage(numValues, numNulls int64, min, max Value) {
+	i.nullPages = append(i.nullPages, numValues == numNulls)
+	i.nullCounts = append(i.nullCounts, numNulls)
+	i.minValues = append(i.minValues, i.class.value(min))
+	i.maxValues = append(i.maxValues, i.class.value(max))
+}
+
+func (i *columnIndexer[T]) ColumnIndex() format.ColumnIndex {
+	minValues := splitFixedLenByteArrayList(sizeof[T](), cast.SliceToBytes(i.minValues))
+	maxValues := splitFixedLenByteArrayList(sizeof[T](), cast.SliceToBytes(i.maxValues))
+	minOrder := i.class.order(i.minValues)
+	maxOrder := i.class.order(i.maxValues)
+	return format.ColumnIndex{
+		NullPages:     i.nullPages,
+		NullCounts:    i.nullCounts,
+		MinValues:     minValues,
+		MaxValues:     maxValues,
+		BoundaryOrder: boundaryOrderOf(minOrder, maxOrder),
+	}
+}
+
+func newBooleanColumnIndexer() *columnIndexer[bool] {
+	return &columnIndexer[bool]{class: &boolClass}
+}
+
+func newInt32ColumnIndexer() *columnIndexer[int32] {
+	return &columnIndexer[int32]{class: &int32Class}
+}
+
+func newInt64ColumnIndexer() *columnIndexer[int64] {
+	return &columnIndexer[int64]{class: &int64Class}
+}
+
+func newInt96ColumnIndexer() *columnIndexer[deprecated.Int96] {
+	return &columnIndexer[deprecated.Int96]{class: &int96Class}
+}
+
+func newFloatColumnIndexer() *columnIndexer[float32] {
+	return &columnIndexer[float32]{class: &float32Class}
+}
+
+func newDoubleColumnIndexer() *columnIndexer[float64] {
+	return &columnIndexer[float64]{class: &float64Class}
+}
+
+func newUint32ColumnIndexer() *columnIndexer[uint32] {
+	return &columnIndexer[uint32]{class: &uint32Class}
+}
+
+func newUint64ColumnIndexer() *columnIndexer[uint64] {
+	return &columnIndexer[uint64]{class: &uint64Class}
+}
diff --git a/column_reader.go b/column_reader.go
index 105d287..a44347a 100644
--- a/column_reader.go
+++ b/column_reader.go
@@ -4,7 +4,6 @@ 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"
@@ -36,7 +35,7 @@ type ColumnReader interface {
 	Reset(decoder encoding.Decoder)
 }
 
-type columnReader struct {
+type fileColumnReader struct {
 	remain             int
 	numValues          int
 	maxRepetitionLevel int8
@@ -46,7 +45,7 @@ type columnReader struct {
 	values             ColumnReader
 }
 
-func newColumnReader(values ColumnReader, maxRepetitionLevel, maxDefinitionLevel int8, bufferSize int) *columnReader {
+func newFileColumnReader(values ColumnReader, maxRepetitionLevel, maxDefinitionLevel int8, bufferSize int) *fileColumnReader {
 	repetitionBufferSize := 0
 	definitionBufferSize := 0
 
@@ -62,7 +61,7 @@ func newColumnReader(values ColumnReader, maxRepetitionLevel, maxDefinitionLevel
 		definitionBufferSize = bufferSize
 	}
 
-	return &columnReader{
+	return &fileColumnReader{
 		maxRepetitionLevel: maxRepetitionLevel,
 		maxDefinitionLevel: maxDefinitionLevel,
 		repetitions:        makeLevelReader(repetitionBufferSize),
@@ -71,11 +70,11 @@ func newColumnReader(values ColumnReader, maxRepetitionLevel, maxDefinitionLevel
 	}
 }
 
-func (r *columnReader) Type() Type { return r.values.Type() }
+func (r *fileColumnReader) Type() Type { return r.values.Type() }
 
-func (r *columnReader) Column() int { return r.values.Column() }
+func (r *fileColumnReader) Column() int { return r.values.Column() }
 
-func (r *columnReader) ReadValues(values []Value) (int, error) {
+func (r *fileColumnReader) ReadValues(values []Value) (int, error) {
 	if r.values == nil {
 		return 0, io.EOF
 	}
@@ -161,7 +160,7 @@ func (r *columnReader) ReadValues(values []Value) (int, error) {
 	return read, nil
 }
 
-func (r *columnReader) reset(numValues int, repetitions, definitions, values encoding.Decoder) {
+func (r *fileColumnReader) reset(numValues int, repetitions, definitions, values encoding.Decoder) {
 	if repetitions != nil {
 		repetitions.SetBitWidth(bits.Len8(r.maxRepetitionLevel))
 	}
@@ -175,8 +174,8 @@ func (r *columnReader) reset(numValues int, repetitions, definitions, values enc
 	r.values.Reset(values)
 }
 
-func (r *columnReader) Reset(encoding.Decoder) {
-	panic("BUG: parquet.columnReader.Reset must not be called")
+func (r *fileColumnReader) Reset(encoding.Decoder) {
+	panic("BUG: parquet.fileColumnReader.Reset must not be called")
 }
 
 type levelReader struct {
@@ -245,458 +244,6 @@ func (r *levelReader) reset(decoder encoding.Decoder) {
 	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
@@ -741,6 +288,10 @@ func (r *byteArrayColumnReader) readByteArrays(do func([]byte) bool) (n int, err
 	}
 }
 
+func (r *byteArrayColumnReader) ReadRequired(values []byte) (int, error) {
+	return r.ReadByteArrays(values)
+}
+
 func (r *byteArrayColumnReader) ReadByteArrays(values []byte) (int, error) {
 	i := 0
 	n, err := r.readByteArrays(func(b []byte) bool {
@@ -800,6 +351,10 @@ func (r *fixedLenByteArrayColumnReader) Type() Type { return r.typ }
 
 func (r *fixedLenByteArrayColumnReader) Column() int { return int(^r.columnIndex) }
 
+func (r *fixedLenByteArrayColumnReader) ReadRequired(values []byte) (int, error) {
+	return r.ReadFixedLenByteArrays(values)
+}
+
 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))
@@ -856,14 +411,3 @@ func (r *fixedLenByteArrayColumnReader) Reset(decoder encoding.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/column_reader_default.go b/column_reader_default.go
new file mode 100644
index 0000000..22b6888
--- /dev/null
+++ b/column_reader_default.go
@@ -0,0 +1,473 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+// 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
+}
+
+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/column_reader_go18.go b/column_reader_go18.go
new file mode 100644
index 0000000..35a94ef
--- /dev/null
+++ b/column_reader_go18.go
@@ -0,0 +1,117 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+type columnReader[T primitive] struct {
+	class       *class[T]
+	typ         Type
+	decoder     encoding.Decoder
+	buffer      []T
+	offset      int
+	bufferSize  int
+	columnIndex int16
+}
+
+func newBooleanColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[bool] {
+	return newColumnReader(typ, columnIndex, bufferSize, &boolClass)
+}
+
+func newInt32ColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[int32] {
+	return newColumnReader(typ, columnIndex, bufferSize, &int32Class)
+}
+
+func newInt64ColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[int64] {
+	return newColumnReader(typ, columnIndex, bufferSize, &int64Class)
+}
+
+func newInt96ColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[deprecated.Int96] {
+	return newColumnReader(typ, columnIndex, bufferSize, &int96Class)
+}
+
+func newFloatColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[float32] {
+	return newColumnReader(typ, columnIndex, bufferSize, &float32Class)
+}
+
+func newDoubleColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[float64] {
+	return newColumnReader(typ, columnIndex, bufferSize, &float64Class)
+}
+
+func newUint32ColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[uint32] {
+	return newColumnReader(typ, columnIndex, bufferSize, &uint32Class)
+}
+
+func newUint64ColumnReader(typ Type, columnIndex int16, bufferSize int) *columnReader[uint64] {
+	return newColumnReader(typ, columnIndex, bufferSize, &uint64Class)
+}
+
+func newColumnReader[T primitive](typ Type, columnIndex int16, bufferSize int, class *class[T]) *columnReader[T] {
+	return &columnReader[T]{
+		class:       class,
+		typ:         typ,
+		bufferSize:  bufferSize,
+		columnIndex: ^columnIndex,
+	}
+}
+
+func (r *columnReader[T]) Type() Type { return r.typ }
+
+func (r *columnReader[T]) Column() int { return int(^r.columnIndex) }
+
+func (r *columnReader[T]) ReadRequired(values []T) (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.class.decode(r.decoder, values)
+	return n + d, err
+}
+
+func (r *columnReader[T]) ReadValues(values []Value) (n int, err error) {
+	if cap(r.buffer) == 0 {
+		r.buffer = make([]T, 0, atLeastOne(r.bufferSize))
+	}
+
+	makeValue := r.class.makeValue
+	columnIndex := r.columnIndex
+	for {
+		for r.offset < len(r.buffer) && n < len(values) {
+			values[n] = makeValue(r.buffer[r.offset])
+			values[n].columnIndex = 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.class.decode(r.decoder, buffer)
+		if d == 0 {
+			return n, err
+		}
+
+		r.buffer = buffer[:d]
+		r.offset = 0
+	}
+}
+
+func (r *columnReader[T]) Reset(decoder encoding.Decoder) {
+	r.decoder = decoder
+	r.buffer = r.buffer[:0]
+	r.offset = 0
+}
diff --git a/convert.go b/convert.go
index 7dae066..dc99524 100644
--- a/convert.go
+++ b/convert.go
@@ -481,8 +481,8 @@ type missingColumnIndex struct{ *missingColumnChunk }
 func (i missingColumnIndex) NumPages() int       { return 1 }
 func (i missingColumnIndex) NullCount(int) int64 { return i.numNulls }
 func (i missingColumnIndex) NullPage(int) bool   { return true }
-func (i missingColumnIndex) MinValue(int) []byte { return nil }
-func (i missingColumnIndex) MaxValue(int) []byte { return nil }
+func (i missingColumnIndex) MinValue(int) Value  { return Value{} }
+func (i missingColumnIndex) MaxValue(int) Value  { return Value{} }
 func (i missingColumnIndex) IsAscending() bool   { return true }
 func (i missingColumnIndex) IsDescending() bool  { return false }
 
diff --git a/dictionary.go b/dictionary.go
index b9b0573..a3a49b7 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -943,13 +943,13 @@ 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 {
+func (index indexedColumnIndex) MinValue(int) Value {
 	min, _ := index.col.Bounds()
-	return min.Bytes()
+	return min
 }
-func (index indexedColumnIndex) MaxValue(int) []byte {
+func (index indexedColumnIndex) MaxValue(int) Value {
 	_, max := index.col.Bounds()
-	return max.Bytes()
+	return max
 }
 func (index indexedColumnIndex) IsAscending() bool {
 	return index.col.typ.Compare(index.col.Bounds()) < 0
diff --git a/encoding/plain/plain_go18.go b/encoding/plain/plain_go18.go
new file mode 100644
index 0000000..cd0e65b
--- /dev/null
+++ b/encoding/plain/plain_go18.go
@@ -0,0 +1,11 @@
+//go:build go1.18
+
+package plain
+
+import "github.com/segmentio/parquet-go/deprecated"
+
+// Type is a type constraint representing the possible Go types which can
+// be represented by the PLAIN encoding.
+type Type interface {
+	bool | int32 | int64 | deprecated.Int96 | float32 | float64 | byte
+}
diff --git a/file.go b/file.go
index ca14841..7035911 100644
--- a/file.go
+++ b/file.go
@@ -423,14 +423,14 @@ func (c *fileColumnChunk) ColumnIndex() ColumnIndex {
 	if c.columnIndex == nil {
 		return nil
 	}
-	return (*columnIndex)(c.columnIndex)
+	return fileColumnIndex{c}
 }
 
 func (c *fileColumnChunk) OffsetIndex() OffsetIndex {
 	if c.offsetIndex == nil {
 		return nil
 	}
-	return (*offsetIndex)(c.offsetIndex)
+	return (*fileOffsetIndex)(c.offsetIndex)
 }
 
 func (c *fileColumnChunk) BloomFilter() BloomFilter {
@@ -935,12 +935,12 @@ func (s *filePageValueReaderState) init(columnType Type, column *Column, codec f
 		}
 		s.reader = columnType.NewColumnReader(int(column.index), bufferSize)
 		if hasLevels {
-			s.reader = newColumnReader(s.reader, maxRepetitionLevel, maxDefinitionLevel, bufferSize)
+			s.reader = newFileColumnReader(s.reader, maxRepetitionLevel, maxDefinitionLevel, bufferSize)
 		}
 	}
 
 	switch r := s.reader.(type) {
-	case *columnReader:
+	case *fileColumnReader:
 		r.reset(int(header.NumValues()), s.repetitions.decoder, s.definitions.decoder, s.page.decoder)
 	default:
 		r.Reset(s.page.decoder)
diff --git a/go.mod b/go.mod
index 2b9c675..605ae34 100644
--- a/go.mod
+++ b/go.mod
@@ -1,12 +1,12 @@
 module github.com/segmentio/parquet-go
 
-go 1.17
+go 1.18
 
 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.13.6
+	github.com/klauspost/compress v1.15.1
 	github.com/pierrec/lz4/v4 v4.1.9
 	github.com/segmentio/encoding v0.3.3
 	golang.org/x/sys v0.0.0-20211110154304-99a53858aa08
diff --git a/go.sum b/go.sum
index 8eb9614..76c8604 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.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
-github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.15.1 h1:y9FcTHGyrebwfP0ZZqFiaxTaiDnUrGkJkI+f583BL1A=
+github.com/klauspost/compress v1.15.1/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/pierrec/lz4/v4 v4.1.9 h1:xkrjwpOP5xg1k4Nn4GX4a4YFGhscyQL/3EddJ1Xxqm8=
 github.com/pierrec/lz4/v4 v4.1.9/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
diff --git a/internal/bits/bits.go b/internal/bits/bits.go
index 7d1df15..a1182b5 100644
--- a/internal/bits/bits.go
+++ b/internal/bits/bits.go
@@ -4,6 +4,8 @@ import (
 	"math/bits"
 )
 
+type uint128 = [16]byte
+
 func BitCount(count int) uint {
 	return 8 * uint(count)
 }
diff --git a/internal/bits/bits_go18.go b/internal/bits/bits_go18.go
new file mode 100644
index 0000000..e0d9baf
--- /dev/null
+++ b/internal/bits/bits_go18.go
@@ -0,0 +1,7 @@
+//go:build go1.18
+
+package bits
+
+type ordered interface {
+	int8 | int16 | int32 | int64 | uint8 | uint16 | uint32 | uint64 | float32 | float64
+}
diff --git a/internal/bits/max_default.go b/internal/bits/max_default.go
index f0c203f..0257722 100644
--- a/internal/bits/max_default.go
+++ b/internal/bits/max_default.go
@@ -1,4 +1,4 @@
-//go:build purego || !amd64
+//go:build !go1.18 && (purego || !amd64)
 
 package bits
 
diff --git a/internal/bits/max_go18.go b/internal/bits/max_go18.go
new file mode 100644
index 0000000..75e73fe
--- /dev/null
+++ b/internal/bits/max_go18.go
@@ -0,0 +1,61 @@
+//go:build go1.18 && (purego || !amd64)
+
+package bits
+
+import (
+	"encoding/binary"
+
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+func maxBool(data []bool) bool {
+	return len(data) > 0 && !boolEqualAll(data, false)
+}
+
+func maxInt32(data []int32) int32 { return max(data) }
+
+func maxInt64(data []int64) int64 { return max(data) }
+
+func maxUint32(data []uint32) uint32 { return max(data) }
+
+func maxUint64(data []uint64) uint64 { return max(data) }
+
+func maxFloat32(data []float32) float32 { return max(data) }
+
+func maxFloat64(data []float64) float64 { return max(data) }
+
+func max[T ordered](data []T) (max T) {
+	if len(data) > 0 {
+		max = data[0]
+
+		for _, value := range data[1:len(data):len(data)] {
+			if value > max {
+				max = value
+			}
+		}
+	}
+	return max
+}
+
+func maxBE128(data []byte) (min []byte) {
+	if len(data) > 0 {
+		be128 := cast.BytesToSlice[uint128](data)
+		m := binary.BigEndian.Uint64(be128[0][:8])
+		j := 0
+		for i := 1; i < len(be128); i++ {
+			x := binary.BigEndian.Uint64(be128[i][:8])
+			switch {
+			case x > m:
+				m, j = x, i
+			case x == m:
+				y := binary.BigEndian.Uint64(be128[i][8:])
+				n := binary.BigEndian.Uint64(be128[j][8:])
+				if y > n {
+					m, j = x, i
+				}
+			}
+		}
+		min = be128[j][:]
+	}
+	return min
+}
diff --git a/internal/bits/min_default.go b/internal/bits/min_default.go
index 55c4ea9..934ca18 100644
--- a/internal/bits/min_default.go
+++ b/internal/bits/min_default.go
@@ -1,4 +1,4 @@
-//go:build purego || !amd64
+//go:build !go1.18 && (purego || !amd64)
 
 package bits
 
diff --git a/internal/bits/min_go18.go b/internal/bits/min_go18.go
new file mode 100644
index 0000000..bd5d33b
--- /dev/null
+++ b/internal/bits/min_go18.go
@@ -0,0 +1,59 @@
+//go:build go1.18 && (purego || !amd64)
+
+package bits
+
+import (
+	"encoding/binary"
+
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+func minBool(data []bool) bool { return boolEqualAll(data, true) }
+
+func minInt32(data []int32) int32 { return min(data) }
+
+func minInt64(data []int64) int64 { return min(data) }
+
+func minUint32(data []uint32) uint32 { return min(data) }
+
+func minUint64(data []uint64) uint64 { return min(data) }
+
+func minFloat32(data []float32) float32 { return min(data) }
+
+func minFloat64(data []float64) float64 { return min(data) }
+
+func min[T ordered](data []T) (min T) {
+	if len(data) > 0 {
+		min = data[0]
+
+		for _, value := range data[1:len(data):len(data)] {
+			if value < min {
+				min = value
+			}
+		}
+	}
+	return min
+}
+
+func minBE128(data []byte) (min []byte) {
+	if len(data) > 0 {
+		be128 := cast.BytesToSlice[uint128](data)
+		m := binary.BigEndian.Uint64(be128[0][:8])
+		j := 0
+		for i := 1; i < len(be128); i++ {
+			x := binary.BigEndian.Uint64(be128[i][:8])
+			switch {
+			case x < m:
+				m, j = x, i
+			case x == m:
+				y := binary.BigEndian.Uint64(be128[i][8:])
+				n := binary.BigEndian.Uint64(be128[j][8:])
+				if y < n {
+					m, j = x, i
+				}
+			}
+		}
+		min = be128[j][:]
+	}
+	return min
+}
diff --git a/internal/bits/minmax_default.go b/internal/bits/minmax_default.go
index eea9cdb..672a326 100644
--- a/internal/bits/minmax_default.go
+++ b/internal/bits/minmax_default.go
@@ -1,4 +1,4 @@
-//go:build purego || !amd64
+//go:build !go1.18 && (purego || !amd64)
 
 package bits
 
diff --git a/internal/bits/minmax_go18.go b/internal/bits/minmax_go18.go
new file mode 100644
index 0000000..e98ba59
--- /dev/null
+++ b/internal/bits/minmax_go18.go
@@ -0,0 +1,96 @@
+//go:build go1.18 && (purego || !amd64)
+
+package bits
+
+import (
+	"encoding/binary"
+
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+func boolEqualAll(data []bool, value bool) bool {
+	for _, v := range data {
+		if v != value {
+			return false
+		}
+	}
+	return len(data) > 0
+}
+
+func minMaxBool(data []bool) (min, max bool) {
+	if len(data) > 0 {
+		switch {
+		case boolEqualAll(data, true):
+			min, max = true, true
+		case boolEqualAll(data, false):
+			min, max = false, false
+		default:
+			min, max = false, true
+		}
+	}
+	return min, max
+}
+
+func minMaxInt32(data []int32) (min, max int32) { return minmax(data) }
+
+func minMaxInt64(data []int64) (min, max int64) { return minmax(data) }
+
+func minMaxUint32(data []uint32) (min, max uint32) { return minmax(data) }
+
+func minMaxUint64(data []uint64) (min, max uint64) { return minmax(data) }
+
+func minMaxFloat32(data []float32) (min, max float32) { return minmax(data) }
+
+func minMaxFloat64(data []float64) (min, max float64) { return minmax(data) }
+
+func minmax[T ordered](data []T) (min, max T) {
+	if len(data) > 0 {
+		min = data[0]
+		max = data[0]
+
+		for _, v := range data[1:len(data):len(data)] {
+			if v < min {
+				min = v
+			}
+			if v > max {
+				max = v
+			}
+		}
+	}
+	return min, max
+}
+
+func minMaxBE128(data []byte) (min, max []byte) {
+	if len(data) > 0 {
+		be128 := cast.BytesToSlice[uint128](data)
+		minHi := binary.BigEndian.Uint64(be128[0][:8])
+		maxHi := minHi
+		minIndex := 0
+		maxIndex := 0
+		for i := 1; i < len(be128); i++ {
+			hi := binary.BigEndian.Uint64(be128[i][:8])
+			lo := binary.BigEndian.Uint64(be128[i][8:])
+			switch {
+			case hi < minHi:
+				minHi, minIndex = hi, i
+			case hi == minHi:
+				minLo := binary.BigEndian.Uint64(be128[minIndex][8:])
+				if lo < minLo {
+					minHi, minIndex = hi, i
+				}
+			}
+			switch {
+			case hi > maxHi:
+				maxHi, maxIndex = hi, i
+			case hi == maxHi:
+				maxLo := binary.BigEndian.Uint64(be128[maxIndex][8:])
+				if lo > maxLo {
+					maxHi, maxIndex = hi, i
+				}
+			}
+		}
+		min = be128[minIndex][:]
+		max = be128[maxIndex][:]
+	}
+	return min, max
+}
diff --git a/internal/bits/order.go b/internal/bits/order.go
index 757f921..ca38dd5 100644
--- a/internal/bits/order.go
+++ b/internal/bits/order.go
@@ -1,6 +1,8 @@
 package bits
 
-import "bytes"
+import (
+	"bytes"
+)
 
 func OrderOfBool(data []bool) int {
 	switch len(data) {
diff --git a/internal/bits/order_default.go b/internal/bits/order_default.go
index fa0e3d0..efe0ac7 100644
--- a/internal/bits/order_default.go
+++ b/internal/bits/order_default.go
@@ -1,4 +1,4 @@
-//go:build purego || !amd64
+//go:build !go1.18 && (purego || !amd64)
 
 package bits
 
diff --git a/internal/bits/order_go18.go b/internal/bits/order_go18.go
new file mode 100644
index 0000000..e308fe7
--- /dev/null
+++ b/internal/bits/order_go18.go
@@ -0,0 +1,43 @@
+//go:build go1.18 && (purego || !amd64)
+
+package bits
+
+func orderOfInt32(data []int32) int { return orderOfValues(data) }
+
+func orderOfInt64(data []int64) int { return orderOfValues(data) }
+
+func orderOfUint32(data []uint32) int { return orderOfValues(data) }
+
+func orderOfUint64(data []uint64) int { return orderOfValues(data) }
+
+func orderOfFloat32(data []float32) int { return orderOfValues(data) }
+
+func orderOfFloat64(data []float64) int { return orderOfValues(data) }
+
+func orderOfValues[T ordered](data []T) int {
+	if valuesAreInAscendingOrder(data) {
+		return +1
+	}
+	if valuesAreInDescendingOrder(data) {
+		return -1
+	}
+	return 0
+}
+
+func valuesAreInAscendingOrder[T ordered](data []T) bool {
+	for i := len(data) - 1; i > 0; i-- {
+		if data[i-1] > data[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func valuesAreInDescendingOrder[T ordered](data []T) bool {
+	for i := len(data) - 1; i > 0; i-- {
+		if data[i-1] < data[i] {
+			return false
+		}
+	}
+	return true
+}
diff --git a/internal/bits/unsafe.go b/internal/bits/unsafe.go
index ee54e13..58c76ff 100644
--- a/internal/bits/unsafe.go
+++ b/internal/bits/unsafe.go
@@ -1,3 +1,5 @@
+//go:build !go1.18
+
 package bits
 
 import "unsafe"
@@ -30,6 +32,10 @@ func Float64ToBytes(data []float64) []byte {
 	return unsafe.Slice(*(**byte)(unsafe.Pointer(&data)), 8*len(data))
 }
 
+func Int16ToUint16(data []int16) []uint16 {
+	return unsafe.Slice(*(**uint16)(unsafe.Pointer(&data)), len(data))
+}
+
 func Int32ToUint32(data []int32) []uint32 {
 	return unsafe.Slice(*(**uint32)(unsafe.Pointer(&data)), len(data))
 }
@@ -38,6 +44,14 @@ func Int64ToUint64(data []int64) []uint64 {
 	return unsafe.Slice(*(**uint64)(unsafe.Pointer(&data)), len(data))
 }
 
+func Float32ToUint32(data []float32) []uint32 {
+	return unsafe.Slice(*(**uint32)(unsafe.Pointer(&data)), len(data))
+}
+
+func Float64ToUint64(data []float64) []uint64 {
+	return unsafe.Slice(*(**uint64)(unsafe.Pointer(&data)), len(data))
+}
+
 func Uint32ToBytes(data []uint32) []byte {
 	return unsafe.Slice(*(**byte)(unsafe.Pointer(&data)), 4*len(data))
 }
diff --git a/internal/bits/unsafe_go18.go b/internal/bits/unsafe_go18.go
new file mode 100644
index 0000000..2361019
--- /dev/null
+++ b/internal/bits/unsafe_go18.go
@@ -0,0 +1,64 @@
+//go:build go1.18
+
+package bits
+
+import "github.com/segmentio/parquet-go/internal/cast"
+
+// TODO: remove these functions and use the internal/cast package instead when
+// we drop support for Go 1.17.
+
+func BoolToBytes(data []bool) []byte { return cast.SliceToBytes(data) }
+
+func Int8ToBytes(data []int8) []byte { return cast.SliceToBytes(data) }
+
+func Int16ToBytes(data []int16) []byte { return cast.SliceToBytes(data) }
+
+func Int32ToBytes(data []int32) []byte { return cast.SliceToBytes(data) }
+
+func Int64ToBytes(data []int64) []byte { return cast.SliceToBytes(data) }
+
+func Float32ToBytes(data []float32) []byte { return cast.SliceToBytes(data) }
+
+func Float64ToBytes(data []float64) []byte { return cast.SliceToBytes(data) }
+
+func Int16ToUint16(data []int16) []uint16 { return cast.Slice[uint16](data) }
+
+func Int32ToUint32(data []int32) []uint32 { return cast.Slice[uint32](data) }
+
+func Int64ToUint64(data []int64) []uint64 { return cast.Slice[uint64](data) }
+
+func Float32ToUint32(data []float32) []uint32 { return cast.Slice[uint32](data) }
+
+func Float64ToUint64(data []float64) []uint64 { return cast.Slice[uint64](data) }
+
+func Uint32ToBytes(data []uint32) []byte { return cast.SliceToBytes(data) }
+
+func Uint64ToBytes(data []uint64) []byte { return cast.SliceToBytes(data) }
+
+func Uint128ToBytes(data [][16]byte) []byte { return cast.SliceToBytes(data) }
+
+func Uint32ToInt32(data []uint32) []int32 { return cast.Slice[int32](data) }
+
+func Uint64ToInt64(data []uint64) []int64 { return cast.Slice[int64](data) }
+
+func BytesToBool(data []byte) []bool { return cast.BytesToSlice[bool](data) }
+
+func BytesToInt8(data []byte) []int8 { return cast.BytesToSlice[int8](data) }
+
+func BytesToInt16(data []byte) []int16 { return cast.BytesToSlice[int16](data) }
+
+func BytesToInt32(data []byte) []int32 { return cast.BytesToSlice[int32](data) }
+
+func BytesToInt64(data []byte) []int64 { return cast.BytesToSlice[int64](data) }
+
+func BytesToUint32(data []byte) []uint32 { return cast.BytesToSlice[uint32](data) }
+
+func BytesToUint64(data []byte) []uint64 { return cast.BytesToSlice[uint64](data) }
+
+func BytesToUint128(data []byte) [][16]byte { return cast.BytesToSlice[uint128](data) }
+
+func BytesToFloat32(data []byte) []float32 { return cast.BytesToSlice[float32](data) }
+
+func BytesToFloat64(data []byte) []float64 { return cast.BytesToSlice[float64](data) }
+
+func BytesToString(data []byte) string { return cast.BytesToString(data) }
diff --git a/internal/cast/cast.go b/internal/cast/cast.go
new file mode 100644
index 0000000..ad8c2e0
--- /dev/null
+++ b/internal/cast/cast.go
@@ -0,0 +1,27 @@
+//go:build go1.18
+
+package cast
+
+import "unsafe"
+
+type Int96 = [3]uint32
+
+type Uint128 = [16]byte
+
+func Slice[To, From any](data []From) []To {
+	var zf From
+	var zt To
+	return unsafe.Slice(*(**To)(unsafe.Pointer(&data)), (uintptr(len(data))*unsafe.Sizeof(zf))/unsafe.Sizeof(zt))
+}
+
+func SliceToBytes[T any](data []T) []byte {
+	return Slice[byte](data)
+}
+
+func BytesToSlice[T any](data []byte) []T {
+	return Slice[T](data)
+}
+
+func BytesToString(data []byte) string {
+	return *(*string)(unsafe.Pointer(&data))
+}
diff --git a/merge.go b/merge.go
index b9b92c5..6849804 100644
--- a/merge.go
+++ b/merge.go
@@ -202,7 +202,7 @@ type rowGroupCursor interface {
 
 type columnSortFunc struct {
 	columnIndex int16
-	compare     sortFunc
+	compare     SortFunc
 }
 
 type bufferedRowGroupCursor struct {
@@ -265,94 +265,3 @@ var (
 	_ RowReaderWithSchema = (*mergedRowGroupRowReader)(nil)
 	//_ RowWriterTo         = (*mergedRowGroupRowReader)(nil)
 )
-
-type sortFunc func(a, b []Value) int
-
-func sortFuncOf(t Type, maxDefinitionLevel, maxRepetitionLevel int8, descending, nullsFirst bool) (sort sortFunc) {
-	switch {
-	case maxRepetitionLevel > 0:
-		sort = sortFuncOfRepeated(t, nullsFirst)
-	case maxDefinitionLevel > 0:
-		sort = sortFuncOfOptional(t, nullsFirst)
-	default:
-		sort = sortFuncOfRequired(t)
-	}
-	if descending {
-		sort = sortFuncOfDescending(sort)
-	}
-	return sort
-}
-
-//go:noinline
-func sortFuncOfDescending(sort sortFunc) sortFunc {
-	return func(a, b []Value) int { return -sort(a, b) }
-}
-
-func sortFuncOfOptional(t Type, nullsFirst bool) sortFunc {
-	if nullsFirst {
-		return sortFuncOfOptionalNullsFirst(t)
-	} else {
-		return sortFuncOfOptionalNullsLast(t)
-	}
-}
-
-//go:noinline
-func sortFuncOfOptionalNullsFirst(t Type) sortFunc {
-	compare := sortFuncOfRequired(t)
-	return func(a, b []Value) int {
-		switch {
-		case a[0].IsNull():
-			if b[0].IsNull() {
-				return 0
-			}
-			return -1
-		case b[0].IsNull():
-			return +1
-		default:
-			return compare(a, b)
-		}
-	}
-}
-
-//go:noinline
-func sortFuncOfOptionalNullsLast(t Type) sortFunc {
-	compare := sortFuncOfRequired(t)
-	return func(a, b []Value) int {
-		switch {
-		case a[0].IsNull():
-			if b[0].IsNull() {
-				return 0
-			}
-			return +1
-		case b[0].IsNull():
-			return -1
-		default:
-			return compare(a, b)
-		}
-	}
-}
-
-//go:noinline
-func sortFuncOfRepeated(t Type, nullsFirst bool) sortFunc {
-	compare := sortFuncOfOptional(t, nullsFirst)
-	return func(a, b []Value) int {
-		n := len(a)
-		if n < len(b) {
-			n = len(b)
-		}
-
-		for i := 0; i < n; i++ {
-			k := compare(a[i:i+1], b[i:i+1])
-			if k != 0 {
-				return k
-			}
-		}
-
-		return len(a) - len(b)
-	}
-}
-
-//go:noinline
-func sortFuncOfRequired(t Type) sortFunc {
-	return func(a, b []Value) int { return t.Compare(a[0], b[0]) }
-}
diff --git a/offset_index.go b/offset_index.go
index 5628d39..a9e5411 100644
--- a/offset_index.go
+++ b/offset_index.go
@@ -23,61 +23,41 @@ type OffsetIndex interface {
 	FirstRowIndex(int) int64
 }
 
-// OffsetIndex is the data structure representing offset indexes.
-type offsetIndex format.OffsetIndex
+type emptyOffsetIndex struct{}
 
-func (index *offsetIndex) NumPages() int {
-	return len(index.PageLocations)
-}
+func (emptyOffsetIndex) NumPages() int                { return 0 }
+func (emptyOffsetIndex) Offset(int) int64             { return 0 }
+func (emptyOffsetIndex) CompressedPageSize(int) int64 { return 0 }
+func (emptyOffsetIndex) FirstRowIndex(int) int64      { return 0 }
 
-func (index *offsetIndex) Offset(i int) int64 {
-	return index.PageLocations[i].Offset
-}
+type fileOffsetIndex format.OffsetIndex
 
-func (index *offsetIndex) CompressedPageSize(i int) int64 {
-	return int64(index.PageLocations[i].CompressedPageSize)
+func (i *fileOffsetIndex) NumPages() int {
+	return len(i.PageLocations)
 }
 
-func (index *offsetIndex) FirstRowIndex(i int) int64 {
-	return index.PageLocations[i].FirstRowIndex
+func (i *fileOffsetIndex) Offset(j int) int64 {
+	return i.PageLocations[j].Offset
 }
 
-func (index booleanPageIndex) Offset(int) int64             { return 0 }
-func (index booleanPageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index booleanPageIndex) FirstRowIndex(int) int64      { return 0 }
-
-func (index int32PageIndex) Offset(int) int64             { return 0 }
-func (index int32PageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index int32PageIndex) FirstRowIndex(int) int64      { return 0 }
-
-func (index int64PageIndex) Offset(int) int64             { return 0 }
-func (index int64PageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index int64PageIndex) FirstRowIndex(int) int64      { return 0 }
-
-func (index int96PageIndex) Offset(int) int64             { return 0 }
-func (index int96PageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index int96PageIndex) FirstRowIndex(int) int64      { return 0 }
-
-func (index floatPageIndex) Offset(int) int64             { return 0 }
-func (index floatPageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index floatPageIndex) FirstRowIndex(int) int64      { return 0 }
+func (i *fileOffsetIndex) CompressedPageSize(j int) int64 {
+	return int64(i.PageLocations[j].CompressedPageSize)
+}
 
-func (index doublePageIndex) Offset(int) int64             { return 0 }
-func (index doublePageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index doublePageIndex) FirstRowIndex(int) int64      { return 0 }
+func (i *fileOffsetIndex) FirstRowIndex(j int) int64 {
+	return i.PageLocations[j].FirstRowIndex
+}
 
-func (index byteArrayPageIndex) Offset(int) int64             { return 0 }
-func (index byteArrayPageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index byteArrayPageIndex) FirstRowIndex(int) int64      { return 0 }
+type byteArrayOffsetIndex struct{ page *byteArrayPage }
 
-func (index fixedLenByteArrayPageIndex) Offset(int) int64             { return 0 }
-func (index fixedLenByteArrayPageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index fixedLenByteArrayPageIndex) FirstRowIndex(int) int64      { return 0 }
+func (i byteArrayOffsetIndex) NumPages() int                { return 1 }
+func (i byteArrayOffsetIndex) Offset(int) int64             { return 0 }
+func (i byteArrayOffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i byteArrayOffsetIndex) FirstRowIndex(int) int64      { return 0 }
 
-func (index uint32PageIndex) Offset(int) int64             { return 0 }
-func (index uint32PageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index uint32PageIndex) FirstRowIndex(int) int64      { return 0 }
+type fixedLenByteArrayOffsetIndex struct{ page *fixedLenByteArrayPage }
 
-func (index uint64PageIndex) Offset(int) int64             { return 0 }
-func (index uint64PageIndex) CompressedPageSize(int) int64 { return index.page.Size() }
-func (index uint64PageIndex) FirstRowIndex(int) int64      { return 0 }
+func (i fixedLenByteArrayOffsetIndex) NumPages() int                { return 1 }
+func (i fixedLenByteArrayOffsetIndex) Offset(int) int64             { return 0 }
+func (i fixedLenByteArrayOffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i fixedLenByteArrayOffsetIndex) FirstRowIndex(int) int64      { return 0 }
diff --git a/offset_index_default.go b/offset_index_default.go
new file mode 100644
index 0000000..71d5166
--- /dev/null
+++ b/offset_index_default.go
@@ -0,0 +1,59 @@
+//go:build !go1.18
+
+package parquet
+
+type booleanOffsetIndex struct{ page *booleanPage }
+
+func (i booleanOffsetIndex) NumPages() int                { return 1 }
+func (i booleanOffsetIndex) Offset(int) int64             { return 0 }
+func (i booleanOffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i booleanOffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type int32OffsetIndex struct{ page *int32Page }
+
+func (i int32OffsetIndex) NumPages() int                { return 1 }
+func (i int32OffsetIndex) Offset(int) int64             { return 0 }
+func (i int32OffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i int32OffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type int64OffsetIndex struct{ page *int64Page }
+
+func (i int64OffsetIndex) NumPages() int                { return 1 }
+func (i int64OffsetIndex) Offset(int) int64             { return 0 }
+func (i int64OffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i int64OffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type int96OffsetIndex struct{ page *int96Page }
+
+func (i int96OffsetIndex) NumPages() int                { return 1 }
+func (i int96OffsetIndex) Offset(int) int64             { return 0 }
+func (i int96OffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i int96OffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type floatOffsetIndex struct{ page *floatPage }
+
+func (i floatOffsetIndex) NumPages() int                { return 1 }
+func (i floatOffsetIndex) Offset(int) int64             { return 0 }
+func (i floatOffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i floatOffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type doubleOffsetIndex struct{ page *doublePage }
+
+func (i doubleOffsetIndex) NumPages() int                { return 1 }
+func (i doubleOffsetIndex) Offset(int) int64             { return 0 }
+func (i doubleOffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i doubleOffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type uint32OffsetIndex struct{ page uint32Page }
+
+func (i uint32OffsetIndex) NumPages() int                { return 1 }
+func (i uint32OffsetIndex) Offset(int) int64             { return 0 }
+func (i uint32OffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i uint32OffsetIndex) FirstRowIndex(int) int64      { return 0 }
+
+type uint64OffsetIndex struct{ page uint64Page }
+
+func (i uint64OffsetIndex) NumPages() int                { return 1 }
+func (i uint64OffsetIndex) Offset(int) int64             { return 0 }
+func (i uint64OffsetIndex) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i uint64OffsetIndex) FirstRowIndex(int) int64      { return 0 }
diff --git a/offset_index_go18.go b/offset_index_go18.go
new file mode 100644
index 0000000..8efdbdd
--- /dev/null
+++ b/offset_index_go18.go
@@ -0,0 +1,10 @@
+//go:build go1.18
+
+package parquet
+
+type offsetIndex[T primitive] struct{ page *page[T] }
+
+func (i offsetIndex[T]) NumPages() int                { return 1 }
+func (i offsetIndex[T]) Offset(int) int64             { return 0 }
+func (i offsetIndex[T]) CompressedPageSize(int) int64 { return i.page.Size() }
+func (i offsetIndex[T]) FirstRowIndex(int) int64      { return 0 }
diff --git a/page.go b/page.go
index 760d327..4c7e77d 100644
--- a/page.go
+++ b/page.go
@@ -546,7 +546,7 @@ func (r *repeatedPageReader) ReadValues(values []Value) (n int, err error) {
 				r.offset++
 				n++
 			}
-			if err != nil {
+			if err != nil && err != io.EOF {
 				return n, err
 			}
 		}
@@ -558,570 +558,6 @@ func (r *repeatedPageReader) ReadValues(values []Value) (n int, err error) {
 	return n, err
 }
 
-type booleanPage struct {
-	values      []bool
-	columnIndex int16
-}
-
-func (page *booleanPage) Column() int { return int(^page.columnIndex) }
-
-func (page *booleanPage) Dictionary() Dictionary { return nil }
-
-func (page *booleanPage) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *booleanPage) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *booleanPage) NumNulls() int64 { return 0 }
-
-func (page *booleanPage) min() bool {
-	for _, value := range page.values {
-		if !value {
-			return false
-		}
-	}
-	return len(page.values) > 0
-}
-
-func (page *booleanPage) max() bool {
-	for _, value := range page.values {
-		if value {
-			return true
-		}
-	}
-	return false
-}
-
-func (page *booleanPage) bounds() (min, max bool) {
-	hasFalse, hasTrue := false, false
-
-	for _, value := range page.values {
-		if value {
-			hasTrue = true
-		} else {
-			hasFalse = true
-		}
-		if hasTrue && hasFalse {
-			break
-		}
-	}
-
-	if !hasFalse {
-		min = true
-	}
-	if hasTrue {
-		max = true
-	}
-	return min, max
-}
-
-func (page *booleanPage) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minBool, maxBool := page.bounds()
-		min = makeValueBoolean(minBool)
-		max = makeValueBoolean(maxBool)
-	}
-	return min, max
-}
-
-func (page *booleanPage) Clone() BufferedPage {
-	return &booleanPage{
-		values:      append([]bool{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *booleanPage) Slice(i, j int64) BufferedPage {
-	return &booleanPage{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *booleanPage) Size() int64 { return sizeOfBool(page.values) }
-
-func (page *booleanPage) RepetitionLevels() []int8 { return nil }
-
-func (page *booleanPage) DefinitionLevels() []int8 { return nil }
-
-func (page *booleanPage) WriteTo(e encoding.Encoder) error { return e.EncodeBoolean(page.values) }
-
-func (page *booleanPage) Values() ValueReader { return &booleanPageReader{page: page} }
-
-func (page *booleanPage) Buffer() BufferedPage { return page }
-
-type booleanPageReader struct {
-	page   *booleanPage
-	offset int
-}
-
-func (r *booleanPageReader) Read(b []byte) (n int, err error) {
-	return r.ReadBooleans(bits.BytesToBool(b))
-}
-
-func (r *booleanPageReader) ReadBooleans(values []bool) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *booleanPageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueBoolean(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-type int32Page struct {
-	values      []int32
-	columnIndex int16
-}
-
-func (page *int32Page) Column() int { return int(^page.columnIndex) }
-
-func (page *int32Page) Dictionary() Dictionary { return nil }
-
-func (page *int32Page) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *int32Page) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *int32Page) NumNulls() int64 { return 0 }
-
-func (page *int32Page) min() int32 { return bits.MinInt32(page.values) }
-
-func (page *int32Page) max() int32 { return bits.MaxInt32(page.values) }
-
-func (page *int32Page) bounds() (min, max int32) { return bits.MinMaxInt32(page.values) }
-
-func (page *int32Page) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minInt32, maxInt32 := page.bounds()
-		min = makeValueInt32(minInt32)
-		max = makeValueInt32(maxInt32)
-	}
-	return min, max
-}
-
-func (page *int32Page) Clone() BufferedPage {
-	return &int32Page{
-		values:      append([]int32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int32Page) Slice(i, j int64) BufferedPage {
-	return &int32Page{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int32Page) Size() int64 { return sizeOfInt32(page.values) }
-
-func (page *int32Page) RepetitionLevels() []int8 { return nil }
-
-func (page *int32Page) DefinitionLevels() []int8 { return nil }
-
-func (page *int32Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt32(page.values) }
-
-func (page *int32Page) Values() ValueReader { return &int32PageReader{page: page} }
-
-func (page *int32Page) Buffer() BufferedPage { return page }
-
-type int32PageReader struct {
-	page   *int32Page
-	offset int
-}
-
-func (r *int32PageReader) Read(b []byte) (n int, err error) {
-	n, err = r.ReadInt32s(bits.BytesToInt32(b))
-	return 4 * n, err
-}
-
-func (r *int32PageReader) ReadInt32s(values []int32) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *int32PageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueInt32(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-type int64Page struct {
-	values      []int64
-	columnIndex int16
-}
-
-func (page *int64Page) Column() int { return int(^page.columnIndex) }
-
-func (page *int64Page) Dictionary() Dictionary { return nil }
-
-func (page *int64Page) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *int64Page) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *int64Page) NumNulls() int64 { return 0 }
-
-func (page *int64Page) min() int64 { return bits.MinInt64(page.values) }
-
-func (page *int64Page) max() int64 { return bits.MaxInt64(page.values) }
-
-func (page *int64Page) bounds() (min, max int64) { return bits.MinMaxInt64(page.values) }
-
-func (page *int64Page) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minInt64, maxInt64 := page.bounds()
-		min = makeValueInt64(minInt64)
-		max = makeValueInt64(maxInt64)
-	}
-	return min, max
-}
-
-func (page *int64Page) Clone() BufferedPage {
-	return &int64Page{
-		values:      append([]int64{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int64Page) Slice(i, j int64) BufferedPage {
-	return &int64Page{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int64Page) Size() int64 { return sizeOfInt64(page.values) }
-
-func (page *int64Page) RepetitionLevels() []int8 { return nil }
-
-func (page *int64Page) DefinitionLevels() []int8 { return nil }
-
-func (page *int64Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt64(page.values) }
-
-func (page *int64Page) Values() ValueReader { return &int64PageReader{page: page} }
-
-func (page *int64Page) Buffer() BufferedPage { return page }
-
-type int64PageReader struct {
-	page   *int64Page
-	offset int
-}
-
-func (r *int64PageReader) Read(b []byte) (n int, err error) {
-	n, err = r.ReadInt64s(bits.BytesToInt64(b))
-	return 8 * n, err
-}
-
-func (r *int64PageReader) ReadInt64s(values []int64) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *int64PageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueInt64(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-type int96Page struct {
-	values      []deprecated.Int96
-	columnIndex int16
-}
-
-func (page *int96Page) Column() int { return int(^page.columnIndex) }
-
-func (page *int96Page) Dictionary() Dictionary { return nil }
-
-func (page *int96Page) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *int96Page) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *int96Page) NumNulls() int64 { return 0 }
-
-func (page *int96Page) min() deprecated.Int96 { return deprecated.MinInt96(page.values) }
-
-func (page *int96Page) max() deprecated.Int96 { return deprecated.MaxInt96(page.values) }
-
-func (page *int96Page) bounds() (min, max deprecated.Int96) {
-	return deprecated.MinMaxInt96(page.values)
-}
-
-func (page *int96Page) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minInt96, maxInt96 := page.bounds()
-		min = makeValueInt96(minInt96)
-		max = makeValueInt96(maxInt96)
-	}
-	return min, max
-}
-
-func (page *int96Page) Clone() BufferedPage {
-	return &int96Page{
-		values:      append([]deprecated.Int96{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int96Page) Slice(i, j int64) BufferedPage {
-	return &int96Page{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *int96Page) Size() int64 { return sizeOfInt96(page.values) }
-
-func (page *int96Page) RepetitionLevels() []int8 { return nil }
-
-func (page *int96Page) DefinitionLevels() []int8 { return nil }
-
-func (page *int96Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt96(page.values) }
-
-func (page *int96Page) Values() ValueReader { return &int96PageReader{page: page} }
-
-func (page *int96Page) Buffer() BufferedPage { return page }
-
-type int96PageReader struct {
-	page   *int96Page
-	offset int
-}
-
-func (r *int96PageReader) Read(b []byte) (n int, err error) {
-	n, err = r.ReadInt96s(deprecated.BytesToInt96(b))
-	return 12 * n, err
-}
-
-func (r *int96PageReader) ReadInt96s(values []deprecated.Int96) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *int96PageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueInt96(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-type floatPage struct {
-	values      []float32
-	columnIndex int16
-}
-
-func (page *floatPage) Column() int { return int(^page.columnIndex) }
-
-func (page *floatPage) Dictionary() Dictionary { return nil }
-
-func (page *floatPage) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *floatPage) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *floatPage) NumNulls() int64 { return 0 }
-
-func (page *floatPage) min() float32 { return bits.MinFloat32(page.values) }
-
-func (page *floatPage) max() float32 { return bits.MaxFloat32(page.values) }
-
-func (page *floatPage) bounds() (min, max float32) { return bits.MinMaxFloat32(page.values) }
-
-func (page *floatPage) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minFloat32, maxFloat32 := page.bounds()
-		min = makeValueFloat(minFloat32)
-		max = makeValueFloat(maxFloat32)
-	}
-	return min, max
-}
-
-func (page *floatPage) Clone() BufferedPage {
-	return &floatPage{
-		values:      append([]float32{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *floatPage) Slice(i, j int64) BufferedPage {
-	return &floatPage{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *floatPage) Size() int64 { return sizeOfFloat32(page.values) }
-
-func (page *floatPage) RepetitionLevels() []int8 { return nil }
-
-func (page *floatPage) DefinitionLevels() []int8 { return nil }
-
-func (page *floatPage) WriteTo(e encoding.Encoder) error { return e.EncodeFloat(page.values) }
-
-func (page *floatPage) Values() ValueReader { return &floatPageReader{page: page} }
-
-func (page *floatPage) Buffer() BufferedPage { return page }
-
-type floatPageReader struct {
-	page   *floatPage
-	offset int
-}
-
-func (r *floatPageReader) Read(b []byte) (n int, err error) {
-	n, err = r.ReadFloats(bits.BytesToFloat32(b))
-	return 4 * n, err
-}
-
-func (r *floatPageReader) ReadFloats(values []float32) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *floatPageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueFloat(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-type doublePage struct {
-	values      []float64
-	columnIndex int16
-}
-
-func (page *doublePage) Column() int { return int(^page.columnIndex) }
-
-func (page *doublePage) Dictionary() Dictionary { return nil }
-
-func (page *doublePage) NumRows() int64 { return int64(len(page.values)) }
-
-func (page *doublePage) NumValues() int64 { return int64(len(page.values)) }
-
-func (page *doublePage) NumNulls() int64 { return 0 }
-
-func (page *doublePage) min() float64 { return bits.MinFloat64(page.values) }
-
-func (page *doublePage) max() float64 { return bits.MaxFloat64(page.values) }
-
-func (page *doublePage) bounds() (min, max float64) { return bits.MinMaxFloat64(page.values) }
-
-func (page *doublePage) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minFloat64, maxFloat64 := page.bounds()
-		min = makeValueDouble(minFloat64)
-		max = makeValueDouble(maxFloat64)
-	}
-	return min, max
-}
-
-func (page *doublePage) Clone() BufferedPage {
-	return &doublePage{
-		values:      append([]float64{}, page.values...),
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *doublePage) Slice(i, j int64) BufferedPage {
-	return &doublePage{
-		values:      page.values[i:j],
-		columnIndex: page.columnIndex,
-	}
-}
-
-func (page *doublePage) Size() int64 { return sizeOfFloat64(page.values) }
-
-func (page *doublePage) RepetitionLevels() []int8 { return nil }
-
-func (page *doublePage) DefinitionLevels() []int8 { return nil }
-
-func (page *doublePage) WriteTo(e encoding.Encoder) error { return e.EncodeDouble(page.values) }
-
-func (page *doublePage) Values() ValueReader { return &doublePageReader{page: page} }
-
-func (page *doublePage) Buffer() BufferedPage { return page }
-
-type doublePageReader struct {
-	page   *doublePage
-	offset int
-}
-
-func (r *doublePageReader) Read(b []byte) (n int, err error) {
-	n, err = r.ReadDoubles(bits.BytesToFloat64(b))
-	return 8 * n, err
-}
-
-func (r *doublePageReader) ReadDoubles(values []float64) (n int, err error) {
-	n = copy(values, r.page.values[r.offset:])
-	r.offset += n
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
-func (r *doublePageReader) ReadValues(values []Value) (n int, err error) {
-	for n < len(values) && r.offset < len(r.page.values) {
-		values[n] = makeValueDouble(r.page.values[r.offset])
-		values[n].columnIndex = r.page.columnIndex
-		r.offset++
-		n++
-	}
-	if r.offset == len(r.page.values) {
-		err = io.EOF
-	}
-	return n, err
-}
-
 type byteArrayPage struct {
 	values      encoding.ByteArrayList
 	columnIndex int16
@@ -1226,6 +662,10 @@ func (r *byteArrayPageReader) Read(b []byte) (int, error) {
 	return n, err
 }
 
+func (r *byteArrayPageReader) ReadRequired(values []byte) (int, error) {
+	return r.ReadByteArrays(values)
+}
+
 func (r *byteArrayPageReader) ReadByteArrays(values []byte) (int, error) {
 	n, _, err := r.readByteArrays(values)
 	return n, err
@@ -1344,6 +784,10 @@ func (r *fixedLenByteArrayPageReader) Read(b []byte) (n int, err error) {
 	return n * r.page.size, err
 }
 
+func (r *fixedLenByteArrayPageReader) ReadRequired(values []byte) (int, error) {
+	return r.ReadFixedLenByteArrays(values)
+}
+
 func (r *fixedLenByteArrayPageReader) ReadFixedLenByteArrays(values []byte) (n int, err error) {
 	n = copy(values, r.page.data[r.offset:]) / r.page.size
 	r.offset += n * r.page.size
@@ -1368,67 +812,6 @@ func (r *fixedLenByteArrayPageReader) ReadValues(values []Value) (n int, err err
 	return n, err
 }
 
-// The following two specializations for unsigned integer types are needed to
-// apply an unsigned comparison when looking up the min and max page values.
-
-type uint32Page struct{ *int32Page }
-
-func (page uint32Page) min() uint32 { return bits.MinUint32(bits.Int32ToUint32(page.values)) }
-
-func (page uint32Page) max() uint32 { return bits.MaxUint32(bits.Int32ToUint32(page.values)) }
-
-func (page uint32Page) bounds() (min, max uint32) {
-	return bits.MinMaxUint32(bits.Int32ToUint32(page.values))
-}
-
-func (page uint32Page) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minUint32, maxUint32 := page.bounds()
-		min = makeValueInt32(int32(minUint32))
-		max = makeValueInt32(int32(maxUint32))
-	}
-	return min, max
-}
-
-func (page uint32Page) Clone() BufferedPage {
-	return uint32Page{page.int32Page.Clone().(*int32Page)}
-}
-
-func (page uint32Page) Slice(i, j int64) BufferedPage {
-	return uint32Page{page.int32Page.Slice(i, j).(*int32Page)}
-}
-
-func (page uint32Page) Buffer() BufferedPage { return page }
-
-type uint64Page struct{ *int64Page }
-
-func (page uint64Page) min() uint64 { return bits.MinUint64(bits.Int64ToUint64(page.values)) }
-
-func (page uint64Page) max() uint64 { return bits.MaxUint64(bits.Int64ToUint64(page.values)) }
-
-func (page uint64Page) bounds() (min, max uint64) {
-	return bits.MinMaxUint64(bits.Int64ToUint64(page.values))
-}
-
-func (page uint64Page) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		minUint64, maxUint64 := page.bounds()
-		min = makeValueInt64(int64(minUint64))
-		max = makeValueInt64(int64(maxUint64))
-	}
-	return min, max
-}
-
-func (page uint64Page) Clone() BufferedPage {
-	return uint64Page{page.int64Page.Clone().(*int64Page)}
-}
-
-func (page uint64Page) Slice(i, j int64) BufferedPage {
-	return uint64Page{page.int64Page.Slice(i, j).(*int64Page)}
-}
-
-func (page uint64Page) Buffer() BufferedPage { return page }
-
 type PageBufferPool interface {
 	GetPageBuffer() io.ReadWriter
 	PutPageBuffer(io.ReadWriter)
@@ -1499,22 +882,4 @@ var (
 
 	_ io.ReaderFrom = (*errorBuffer)(nil)
 	_ io.WriterTo   = (*errorBuffer)(nil)
-
-	_ io.Reader = (*booleanPageReader)(nil)
-	_ io.Reader = (*int32PageReader)(nil)
-	_ io.Reader = (*int64PageReader)(nil)
-	_ io.Reader = (*int96PageReader)(nil)
-	_ io.Reader = (*floatPageReader)(nil)
-	_ io.Reader = (*doublePageReader)(nil)
-	_ io.Reader = (*byteArrayPageReader)(nil)
-	_ io.Reader = (*fixedLenByteArrayPageReader)(nil)
-
-	_ BooleanReader           = (*booleanPageReader)(nil)
-	_ Int32Reader             = (*int32PageReader)(nil)
-	_ Int64Reader             = (*int64PageReader)(nil)
-	_ Int96Reader             = (*int96PageReader)(nil)
-	_ FloatReader             = (*floatPageReader)(nil)
-	_ DoubleReader            = (*doublePageReader)(nil)
-	_ ByteArrayReader         = (*byteArrayPageReader)(nil)
-	_ FixedLenByteArrayReader = (*fixedLenByteArrayPageReader)(nil)
 )
diff --git a/page_default.go b/page_default.go
new file mode 100644
index 0000000..caa96c2
--- /dev/null
+++ b/page_default.go
@@ -0,0 +1,656 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+type booleanPage struct {
+	values      []bool
+	columnIndex int16
+}
+
+func (page *booleanPage) Column() int { return int(^page.columnIndex) }
+
+func (page *booleanPage) Dictionary() Dictionary { return nil }
+
+func (page *booleanPage) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *booleanPage) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *booleanPage) NumNulls() int64 { return 0 }
+
+func (page *booleanPage) min() bool {
+	for _, value := range page.values {
+		if !value {
+			return false
+		}
+	}
+	return len(page.values) > 0
+}
+
+func (page *booleanPage) max() bool {
+	for _, value := range page.values {
+		if value {
+			return true
+		}
+	}
+	return false
+}
+
+func (page *booleanPage) bounds() (min, max bool) {
+	hasFalse, hasTrue := false, false
+
+	for _, value := range page.values {
+		if value {
+			hasTrue = true
+		} else {
+			hasFalse = true
+		}
+		if hasTrue && hasFalse {
+			break
+		}
+	}
+
+	if !hasFalse {
+		min = true
+	}
+	if hasTrue {
+		max = true
+	}
+	return min, max
+}
+
+func (page *booleanPage) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minBool, maxBool := page.bounds()
+		min = makeValueBoolean(minBool)
+		max = makeValueBoolean(maxBool)
+	}
+	return min, max
+}
+
+func (page *booleanPage) Clone() BufferedPage {
+	return &booleanPage{
+		values:      append([]bool{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *booleanPage) Slice(i, j int64) BufferedPage {
+	return &booleanPage{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *booleanPage) Size() int64 { return sizeOfBool(page.values) }
+
+func (page *booleanPage) RepetitionLevels() []int8 { return nil }
+
+func (page *booleanPage) DefinitionLevels() []int8 { return nil }
+
+func (page *booleanPage) WriteTo(e encoding.Encoder) error { return e.EncodeBoolean(page.values) }
+
+func (page *booleanPage) Values() ValueReader { return &booleanPageReader{page: page} }
+
+func (page *booleanPage) Buffer() BufferedPage { return page }
+
+type booleanPageReader struct {
+	page   *booleanPage
+	offset int
+}
+
+func (r *booleanPageReader) Read(b []byte) (n int, err error) {
+	return r.ReadBooleans(bits.BytesToBool(b))
+}
+
+func (r *booleanPageReader) ReadBooleans(values []bool) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *booleanPageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueBoolean(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type int32Page struct {
+	values      []int32
+	columnIndex int16
+}
+
+func (page *int32Page) Column() int { return int(^page.columnIndex) }
+
+func (page *int32Page) Dictionary() Dictionary { return nil }
+
+func (page *int32Page) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *int32Page) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *int32Page) NumNulls() int64 { return 0 }
+
+func (page *int32Page) min() int32 { return bits.MinInt32(page.values) }
+
+func (page *int32Page) max() int32 { return bits.MaxInt32(page.values) }
+
+func (page *int32Page) bounds() (min, max int32) { return bits.MinMaxInt32(page.values) }
+
+func (page *int32Page) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minInt32, maxInt32 := page.bounds()
+		min = makeValueInt32(minInt32)
+		max = makeValueInt32(maxInt32)
+	}
+	return min, max
+}
+
+func (page *int32Page) Clone() BufferedPage {
+	return &int32Page{
+		values:      append([]int32{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int32Page) Slice(i, j int64) BufferedPage {
+	return &int32Page{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int32Page) Size() int64 { return sizeOfInt32(page.values) }
+
+func (page *int32Page) RepetitionLevels() []int8 { return nil }
+
+func (page *int32Page) DefinitionLevels() []int8 { return nil }
+
+func (page *int32Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt32(page.values) }
+
+func (page *int32Page) Values() ValueReader { return &int32PageReader{page: page} }
+
+func (page *int32Page) Buffer() BufferedPage { return page }
+
+type int32PageReader struct {
+	page   *int32Page
+	offset int
+}
+
+func (r *int32PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt32s(bits.BytesToInt32(b))
+	return 4 * n, err
+}
+
+func (r *int32PageReader) ReadInt32s(values []int32) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *int32PageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueInt32(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type int64Page struct {
+	values      []int64
+	columnIndex int16
+}
+
+func (page *int64Page) Column() int { return int(^page.columnIndex) }
+
+func (page *int64Page) Dictionary() Dictionary { return nil }
+
+func (page *int64Page) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *int64Page) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *int64Page) NumNulls() int64 { return 0 }
+
+func (page *int64Page) min() int64 { return bits.MinInt64(page.values) }
+
+func (page *int64Page) max() int64 { return bits.MaxInt64(page.values) }
+
+func (page *int64Page) bounds() (min, max int64) { return bits.MinMaxInt64(page.values) }
+
+func (page *int64Page) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minInt64, maxInt64 := page.bounds()
+		min = makeValueInt64(minInt64)
+		max = makeValueInt64(maxInt64)
+	}
+	return min, max
+}
+
+func (page *int64Page) Clone() BufferedPage {
+	return &int64Page{
+		values:      append([]int64{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int64Page) Slice(i, j int64) BufferedPage {
+	return &int64Page{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int64Page) Size() int64 { return sizeOfInt64(page.values) }
+
+func (page *int64Page) RepetitionLevels() []int8 { return nil }
+
+func (page *int64Page) DefinitionLevels() []int8 { return nil }
+
+func (page *int64Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt64(page.values) }
+
+func (page *int64Page) Values() ValueReader { return &int64PageReader{page: page} }
+
+func (page *int64Page) Buffer() BufferedPage { return page }
+
+type int64PageReader struct {
+	page   *int64Page
+	offset int
+}
+
+func (r *int64PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt64s(bits.BytesToInt64(b))
+	return 8 * n, err
+}
+
+func (r *int64PageReader) ReadInt64s(values []int64) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *int64PageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueInt64(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type int96Page struct {
+	values      []deprecated.Int96
+	columnIndex int16
+}
+
+func (page *int96Page) Column() int { return int(^page.columnIndex) }
+
+func (page *int96Page) Dictionary() Dictionary { return nil }
+
+func (page *int96Page) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *int96Page) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *int96Page) NumNulls() int64 { return 0 }
+
+func (page *int96Page) min() deprecated.Int96 { return deprecated.MinInt96(page.values) }
+
+func (page *int96Page) max() deprecated.Int96 { return deprecated.MaxInt96(page.values) }
+
+func (page *int96Page) bounds() (min, max deprecated.Int96) {
+	return deprecated.MinMaxInt96(page.values)
+}
+
+func (page *int96Page) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minInt96, maxInt96 := page.bounds()
+		min = makeValueInt96(minInt96)
+		max = makeValueInt96(maxInt96)
+	}
+	return min, max
+}
+
+func (page *int96Page) Clone() BufferedPage {
+	return &int96Page{
+		values:      append([]deprecated.Int96{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int96Page) Slice(i, j int64) BufferedPage {
+	return &int96Page{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *int96Page) Size() int64 { return sizeOfInt96(page.values) }
+
+func (page *int96Page) RepetitionLevels() []int8 { return nil }
+
+func (page *int96Page) DefinitionLevels() []int8 { return nil }
+
+func (page *int96Page) WriteTo(e encoding.Encoder) error { return e.EncodeInt96(page.values) }
+
+func (page *int96Page) Values() ValueReader { return &int96PageReader{page: page} }
+
+func (page *int96Page) Buffer() BufferedPage { return page }
+
+type int96PageReader struct {
+	page   *int96Page
+	offset int
+}
+
+func (r *int96PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt96s(deprecated.BytesToInt96(b))
+	return 12 * n, err
+}
+
+func (r *int96PageReader) ReadInt96s(values []deprecated.Int96) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *int96PageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueInt96(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type floatPage struct {
+	values      []float32
+	columnIndex int16
+}
+
+func (page *floatPage) Column() int { return int(^page.columnIndex) }
+
+func (page *floatPage) Dictionary() Dictionary { return nil }
+
+func (page *floatPage) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *floatPage) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *floatPage) NumNulls() int64 { return 0 }
+
+func (page *floatPage) min() float32 { return bits.MinFloat32(page.values) }
+
+func (page *floatPage) max() float32 { return bits.MaxFloat32(page.values) }
+
+func (page *floatPage) bounds() (min, max float32) { return bits.MinMaxFloat32(page.values) }
+
+func (page *floatPage) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minFloat32, maxFloat32 := page.bounds()
+		min = makeValueFloat(minFloat32)
+		max = makeValueFloat(maxFloat32)
+	}
+	return min, max
+}
+
+func (page *floatPage) Clone() BufferedPage {
+	return &floatPage{
+		values:      append([]float32{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *floatPage) Slice(i, j int64) BufferedPage {
+	return &floatPage{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *floatPage) Size() int64 { return sizeOfFloat32(page.values) }
+
+func (page *floatPage) RepetitionLevels() []int8 { return nil }
+
+func (page *floatPage) DefinitionLevels() []int8 { return nil }
+
+func (page *floatPage) WriteTo(e encoding.Encoder) error { return e.EncodeFloat(page.values) }
+
+func (page *floatPage) Values() ValueReader { return &floatPageReader{page: page} }
+
+func (page *floatPage) Buffer() BufferedPage { return page }
+
+type floatPageReader struct {
+	page   *floatPage
+	offset int
+}
+
+func (r *floatPageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadFloats(bits.BytesToFloat32(b))
+	return 4 * n, err
+}
+
+func (r *floatPageReader) ReadFloats(values []float32) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *floatPageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueFloat(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type doublePage struct {
+	values      []float64
+	columnIndex int16
+}
+
+func (page *doublePage) Column() int { return int(^page.columnIndex) }
+
+func (page *doublePage) Dictionary() Dictionary { return nil }
+
+func (page *doublePage) NumRows() int64 { return int64(len(page.values)) }
+
+func (page *doublePage) NumValues() int64 { return int64(len(page.values)) }
+
+func (page *doublePage) NumNulls() int64 { return 0 }
+
+func (page *doublePage) min() float64 { return bits.MinFloat64(page.values) }
+
+func (page *doublePage) max() float64 { return bits.MaxFloat64(page.values) }
+
+func (page *doublePage) bounds() (min, max float64) { return bits.MinMaxFloat64(page.values) }
+
+func (page *doublePage) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minFloat64, maxFloat64 := page.bounds()
+		min = makeValueDouble(minFloat64)
+		max = makeValueDouble(maxFloat64)
+	}
+	return min, max
+}
+
+func (page *doublePage) Clone() BufferedPage {
+	return &doublePage{
+		values:      append([]float64{}, page.values...),
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *doublePage) Slice(i, j int64) BufferedPage {
+	return &doublePage{
+		values:      page.values[i:j],
+		columnIndex: page.columnIndex,
+	}
+}
+
+func (page *doublePage) Size() int64 { return sizeOfFloat64(page.values) }
+
+func (page *doublePage) RepetitionLevels() []int8 { return nil }
+
+func (page *doublePage) DefinitionLevels() []int8 { return nil }
+
+func (page *doublePage) WriteTo(e encoding.Encoder) error { return e.EncodeDouble(page.values) }
+
+func (page *doublePage) Values() ValueReader { return &doublePageReader{page: page} }
+
+func (page *doublePage) Buffer() BufferedPage { return page }
+
+type doublePageReader struct {
+	page   *doublePage
+	offset int
+}
+
+func (r *doublePageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadDoubles(bits.BytesToFloat64(b))
+	return 8 * n, err
+}
+
+func (r *doublePageReader) ReadDoubles(values []float64) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *doublePageReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.offset < len(r.page.values) {
+		values[n] = makeValueDouble(r.page.values[r.offset])
+		values[n].columnIndex = r.page.columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+// The following two specializations for unsigned integer types are needed to
+// apply an unsigned comparison when looking up the min and max page values.
+
+type uint32Page struct{ *int32Page }
+
+func (page uint32Page) min() uint32 { return bits.MinUint32(bits.Int32ToUint32(page.values)) }
+
+func (page uint32Page) max() uint32 { return bits.MaxUint32(bits.Int32ToUint32(page.values)) }
+
+func (page uint32Page) bounds() (min, max uint32) {
+	return bits.MinMaxUint32(bits.Int32ToUint32(page.values))
+}
+
+func (page uint32Page) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minUint32, maxUint32 := page.bounds()
+		min = makeValueInt32(int32(minUint32))
+		max = makeValueInt32(int32(maxUint32))
+	}
+	return min, max
+}
+
+func (page uint32Page) Clone() BufferedPage {
+	return uint32Page{page.int32Page.Clone().(*int32Page)}
+}
+
+func (page uint32Page) Slice(i, j int64) BufferedPage {
+	return uint32Page{page.int32Page.Slice(i, j).(*int32Page)}
+}
+
+func (page uint32Page) Buffer() BufferedPage { return page }
+
+type uint64Page struct{ *int64Page }
+
+func (page uint64Page) min() uint64 { return bits.MinUint64(bits.Int64ToUint64(page.values)) }
+
+func (page uint64Page) max() uint64 { return bits.MaxUint64(bits.Int64ToUint64(page.values)) }
+
+func (page uint64Page) bounds() (min, max uint64) {
+	return bits.MinMaxUint64(bits.Int64ToUint64(page.values))
+}
+
+func (page uint64Page) Bounds() (min, max Value) {
+	if len(page.values) > 0 {
+		minUint64, maxUint64 := page.bounds()
+		min = makeValueInt64(int64(minUint64))
+		max = makeValueInt64(int64(maxUint64))
+	}
+	return min, max
+}
+
+func (page uint64Page) Clone() BufferedPage {
+	return uint64Page{page.int64Page.Clone().(*int64Page)}
+}
+
+func (page uint64Page) Slice(i, j int64) BufferedPage {
+	return uint64Page{page.int64Page.Slice(i, j).(*int64Page)}
+}
+
+func (page uint64Page) Buffer() BufferedPage { return page }
+
+var (
+	_ io.Reader = (*booleanPageReader)(nil)
+	_ io.Reader = (*int32PageReader)(nil)
+	_ io.Reader = (*int64PageReader)(nil)
+	_ io.Reader = (*int96PageReader)(nil)
+	_ io.Reader = (*floatPageReader)(nil)
+	_ io.Reader = (*doublePageReader)(nil)
+	_ io.Reader = (*byteArrayPageReader)(nil)
+	_ io.Reader = (*fixedLenByteArrayPageReader)(nil)
+
+	_ BooleanReader           = (*booleanPageReader)(nil)
+	_ Int32Reader             = (*int32PageReader)(nil)
+	_ Int64Reader             = (*int64PageReader)(nil)
+	_ Int96Reader             = (*int96PageReader)(nil)
+	_ FloatReader             = (*floatPageReader)(nil)
+	_ DoubleReader            = (*doublePageReader)(nil)
+	_ ByteArrayReader         = (*byteArrayPageReader)(nil)
+	_ FixedLenByteArrayReader = (*fixedLenByteArrayPageReader)(nil)
+)
diff --git a/page_go18.go b/page_go18.go
new file mode 100644
index 0000000..b398994
--- /dev/null
+++ b/page_go18.go
@@ -0,0 +1,107 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"io"
+
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+type page[T primitive] struct {
+	class       *class[T]
+	values      []T
+	columnIndex int16
+}
+
+func (p *page[T]) Column() int { return int(^p.columnIndex) }
+
+func (p *page[T]) Dictionary() Dictionary { return nil }
+
+func (p *page[T]) NumRows() int64 { return int64(len(p.values)) }
+
+func (p *page[T]) NumValues() int64 { return int64(len(p.values)) }
+
+func (p *page[T]) NumNulls() int64 { return 0 }
+
+func (p *page[T]) min() T { return p.class.min(p.values) }
+
+func (p *page[T]) max() T { return p.class.max(p.values) }
+
+func (p *page[T]) bounds() (T, T) { return p.class.bounds(p.values) }
+
+func (p *page[T]) Bounds() (min, max Value) {
+	if len(p.values) > 0 {
+		minValue, maxValue := p.bounds()
+		min = p.class.makeValue(minValue)
+		max = p.class.makeValue(maxValue)
+	}
+	return min, max
+}
+
+func (p *page[T]) Clone() BufferedPage {
+	return &page[T]{
+		class:       p.class,
+		values:      append([]T{}, p.values...),
+		columnIndex: p.columnIndex,
+	}
+}
+
+func (p *page[T]) Slice(i, j int64) BufferedPage {
+	return &page[T]{
+		class:       p.class,
+		values:      p.values[i:j],
+		columnIndex: p.columnIndex,
+	}
+}
+
+func (p *page[T]) Size() int64 { return int64(len(p.values)) * int64(sizeof[T]()) }
+
+func (p *page[T]) RepetitionLevels() []int8 { return nil }
+
+func (p *page[T]) DefinitionLevels() []int8 { return nil }
+
+func (p *page[T]) WriteTo(e encoding.Encoder) error { return p.class.encode(e, p.values) }
+
+func (p *page[T]) Values() ValueReader { return &pageReader[T]{page: p} }
+
+func (p *page[T]) Buffer() BufferedPage { return p }
+
+type pageReader[T primitive] struct {
+	page   *page[T]
+	offset int
+}
+
+func (r *pageReader[T]) Read(b []byte) (n int, err error) {
+	return r.ReadRequired(cast.BytesToSlice[T](b))
+}
+
+func (r *pageReader[T]) ReadRequired(values []T) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+func (r *pageReader[T]) ReadValues(values []Value) (n int, err error) {
+	makeValue := r.page.class.makeValue
+	pageValues := r.page.values
+	columnIndex := r.page.columnIndex
+	for n < len(values) && r.offset < len(pageValues) {
+		values[n] = makeValue(pageValues[r.offset])
+		values[n].columnIndex = columnIndex
+		r.offset++
+		n++
+	}
+	if r.offset == len(pageValues) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+var (
+	_ RequiredReader[bool] = (*pageReader[bool])(nil)
+)
diff --git a/row_group.go b/row_group.go
index 18b95d3..ab1baeb 100644
--- a/row_group.go
+++ b/row_group.go
@@ -210,10 +210,12 @@ func MergeRowGroups(rowGroups []RowGroup, options ...RowGroupOption) (RowGroup,
 				columnIndex: leaf.columnIndex,
 				compare: sortFuncOf(
 					leaf.node.Type(),
-					leaf.maxRepetitionLevel,
-					leaf.maxDefinitionLevel,
-					m.sorting[sortingIndex].Descending(),
-					m.sorting[sortingIndex].NullsFirst(),
+					&SortConfig{
+						MaxRepetitionLevel: int(leaf.maxRepetitionLevel),
+						MaxDefinitionLevel: int(leaf.maxDefinitionLevel),
+						Descending:         m.sorting[sortingIndex].Descending(),
+						NullsFirst:         m.sorting[sortingIndex].NullsFirst(),
+					},
 				),
 			}
 		}
@@ -447,8 +449,8 @@ type emptyColumnChunk struct {
 func (c *emptyColumnChunk) Type() Type               { return c.typ }
 func (c *emptyColumnChunk) Column() int              { return int(c.column) }
 func (c *emptyColumnChunk) Pages() Pages             { return emptyPages{} }
-func (c *emptyColumnChunk) ColumnIndex() ColumnIndex { return &emptyColumnIndex }
-func (c *emptyColumnChunk) OffsetIndex() OffsetIndex { return &emptyOffsetIndex }
+func (c *emptyColumnChunk) ColumnIndex() ColumnIndex { return emptyColumnIndex{} }
+func (c *emptyColumnChunk) OffsetIndex() OffsetIndex { return emptyOffsetIndex{} }
 func (c *emptyColumnChunk) BloomFilter() BloomFilter { return emptyBloomFilter{} }
 func (c *emptyColumnChunk) NumValues() int64         { return 0 }
 
@@ -471,9 +473,6 @@ func (emptyPages) ReadPage() (Page, error) { return nil, io.EOF }
 func (emptyPages) SeekToRow(int64) error   { return nil }
 
 var (
-	emptyColumnIndex = columnIndex{}
-	emptyOffsetIndex = offsetIndex{}
-
 	_ RowReaderWithSchema = (*rowGroupRowReader)(nil)
 	_ RowWriterTo         = (*rowGroupRowReader)(nil)
 
diff --git a/sort.go b/sort.go
new file mode 100644
index 0000000..fdfff59
--- /dev/null
+++ b/sort.go
@@ -0,0 +1,182 @@
+package parquet
+
+// The SortConfig type carries configuration options used to generate sorting
+// functions.
+//
+// SortConfig implements the SortOption interface so it can be used directly as
+// argument to the SortFuncOf function, for example:
+//
+//	sortFunc := parquet.SortFuncOf(columnType, &parquet.SortConfig{
+//		Descending: true,
+//		NullsFirst: true,
+//	})
+//
+type SortConfig struct {
+	MaxRepetitionLevel int
+	MaxDefinitionLevel int
+	Descending         bool
+	NullsFirst         bool
+}
+
+// Apply applies options to c.
+func (c *SortConfig) Apply(options ...SortOption) {
+	for _, opt := range options {
+		opt.ConfigureSort(c)
+	}
+}
+
+// ConfigureSort satisfies the SortOption interface.
+func (c *SortConfig) ConfigureSort(config *SortConfig) {
+	*c = *config
+}
+
+// SortMaxRepetitionLevel constructs a configuration option which sets the
+// maximum repetition level known to a sorting function.
+//
+// Defaults to zero, which represents a non-repeated column.
+func SortMaxRepetitionLevel(level int) SortOption {
+	return sortingOption(func(c *SortConfig) { c.MaxRepetitionLevel = level })
+}
+
+// SortMaxDefinitionLevel constructs a configuration option which sets the
+// maximum definition level known to a sorting function.
+//
+// Defaults to zero, which represents a non-nullable column.
+func SortMaxDefinitionLevel(level int) SortOption {
+	return sortingOption(func(c *SortConfig) { c.MaxDefinitionLevel = level })
+}
+
+// SortDescending constructs a configuration option which inverts the order of a
+// sorting function.
+//
+// Defaults to false, which means values are sorted in ascending order.
+func SortDescending(descending bool) SortOption {
+	return sortingOption(func(c *SortConfig) { c.Descending = descending })
+}
+
+// SortNullsFirst constructs a configuration option which places the null values
+// first or last.
+//
+// Defaults to false, which means null values are placed last.
+func SortNullsFirst(nullsFirst bool) SortOption {
+	return sortingOption(func(c *SortConfig) { c.NullsFirst = nullsFirst })
+}
+
+// SortOption is an interface implemented by types that carry configuration
+// options for sorting functions.
+type SortOption interface {
+	ConfigureSort(*SortConfig)
+}
+
+type sortingOption func(*SortConfig)
+
+func (f sortingOption) ConfigureSort(c *SortConfig) { f(c) }
+
+// SortFunc is a function type which compares two sets of column values.
+//
+// Slices with exactly one value must be passed to the function when comparing
+// values of non-repeated columns. For repeated columns, there may be zero or
+// more values in each slice, and the parameters may have different lengths.
+//
+// SortFunc is a low-level API which is usually useful to construct customize
+// implementations of the RowGroup interface.
+type SortFunc func(a, b []Value) int
+
+// SortFuncOf constructs a sorting function for values of the given type.
+//
+// The list of options contains the configuration used to construct the sorting
+// function.
+func SortFuncOf(t Type, options ...SortOption) SortFunc {
+	config := new(SortConfig)
+	config.Apply(options...)
+	return sortFuncOf(t, config)
+}
+
+func sortFuncOf(t Type, config *SortConfig) (sort SortFunc) {
+	sort = sortFuncOfRequired(t)
+
+	if config.Descending {
+		sort = sortFuncOfDescending(sort)
+	}
+
+	switch {
+	case makeRepetitionLevel(config.MaxRepetitionLevel) > 0:
+		sort = sortFuncOfRepeated(sort, config)
+	case makeDefinitionLevel(config.MaxDefinitionLevel) > 0:
+		sort = sortFuncOfOptional(sort, config)
+	}
+
+	return sort
+}
+
+//go:noinline
+func sortFuncOfDescending(sort SortFunc) SortFunc {
+	return func(a, b []Value) int { return -sort(a, b) }
+}
+
+func sortFuncOfOptional(sort SortFunc, config *SortConfig) SortFunc {
+	if config.NullsFirst {
+		return sortFuncOfOptionalNullsFirst(sort)
+	} else {
+		return sortFuncOfOptionalNullsLast(sort)
+	}
+}
+
+//go:noinline
+func sortFuncOfOptionalNullsFirst(sort SortFunc) SortFunc {
+	return func(a, b []Value) int {
+		switch {
+		case a[0].IsNull():
+			if b[0].IsNull() {
+				return 0
+			}
+			return -1
+		case b[0].IsNull():
+			return +1
+		default:
+			return sort(a, b)
+		}
+	}
+}
+
+//go:noinline
+func sortFuncOfOptionalNullsLast(sort SortFunc) SortFunc {
+	return func(a, b []Value) int {
+		switch {
+		case a[0].IsNull():
+			if b[0].IsNull() {
+				return 0
+			}
+			return +1
+		case b[0].IsNull():
+			return -1
+		default:
+			return sort(a, b)
+		}
+	}
+}
+
+//go:noinline
+func sortFuncOfRepeated(sort SortFunc, config *SortConfig) SortFunc {
+	sort = sortFuncOfOptional(sort, config)
+	return func(a, b []Value) int {
+		n := len(a)
+		if n > len(b) {
+			n = len(b)
+		}
+
+		for i := 0; i < n; i++ {
+			k := sort(a[i:i+1], b[i:i+1])
+			if k != 0 {
+				return k
+			}
+		}
+
+		return len(a) - len(b)
+	}
+}
+
+//go:noinline
+func sortFuncOfRequired(t Type) SortFunc {
+	return func(a, b []Value) int { return t.Compare(a[0], b[0]) }
+}
diff --git a/type.go b/type.go
index b95a53d..298f380 100644
--- a/type.go
+++ b/type.go
@@ -30,6 +30,18 @@ const (
 // String returns a human-readable representation of the physical type.
 func (k Kind) String() string { return format.Type(k).String() }
 
+// Value constructs a value form k and v.
+//
+// The method panics if the data is not a valid representation of the value
+// kind; for example, if the kind is Int32 but the data is not 4 bytes long.
+func (k Kind) Value(v []byte) Value {
+	x, err := parseValue(k, v)
+	if err != nil {
+		panic(err)
+	}
+	return x
+}
+
 // The Type interface represents logical types of the parquet type system.
 //
 // Types are immutable and therefore safe to access from multiple goroutines.
diff --git a/value.go b/value.go
index 807cbb1..c600d8d 100644
--- a/value.go
+++ b/value.go
@@ -119,176 +119,6 @@ func copyValues(dst ValueWriter, src ValueReader, buf []Value) (written int64, e
 	}
 }
 
-// BooleanReader is an interface implemented by ValueReader instances which
-// expose the content of a column of boolean values.
-type BooleanReader interface {
-	// Read boolean values into the buffer passed as argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadBooleans(values []bool) (int, error)
-}
-
-// BooleanWriter is an interface implemented by ValueWriter instances which
-// support writing columns of boolean values.
-type BooleanWriter interface {
-	// Write boolean values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteBooleans(values []bool) (int, error)
-}
-
-// Int32Reader is an interface implemented by ValueReader instances which expose
-// the content of a column of int32 values.
-type Int32Reader interface {
-	// Read 32 bits integer values into the buffer passed as argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadInt32s(values []int32) (int, error)
-}
-
-// Int32Writer is an interface implemented by ValueWriter instances which
-// support writing columns of 32 bits signed integer values.
-type Int32Writer interface {
-	// Write 32 bits signed integer values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteInt32s(values []int32) (int, error)
-}
-
-// Int64Reader is an interface implemented by ValueReader instances which expose
-// the content of a column of int64 values.
-type Int64Reader interface {
-	// Read 64 bits integer values into the buffer passed as argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadInt64s(values []int64) (int, error)
-}
-
-// Int64Writer is an interface implemented by ValueWriter instances which
-// support writing columns of 64 bits signed integer values.
-type Int64Writer interface {
-	// Write 64 bits signed integer values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteInt64s(values []int64) (int, error)
-}
-
-// Int96Reader is an interface implemented by ValueReader instances which expose
-// the content of a column of int96 values.
-type Int96Reader interface {
-	// Read 96 bits integer values into the buffer passed as argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadInt96s(values []deprecated.Int96) (int, error)
-}
-
-// Int96Writer is an interface implemented by ValueWriter instances which
-// support writing columns of 96 bits signed integer values.
-type Int96Writer interface {
-	// Write 96 bits signed integer values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteInt96s(values []deprecated.Int96) (int, error)
-}
-
-// FloatReader is an interface implemented by ValueReader instances which expose
-// the content of a column of single-precision floating point values.
-type FloatReader interface {
-	// Read single-precision floating point values into the buffer passed as
-	// argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadFloats(values []float32) (int, error)
-}
-
-// FloatWriter is an interface implemented by ValueWriter instances which
-// support writing columns of single-precision floating point values.
-type FloatWriter interface {
-	// Write single-precision floating point values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteFloats(values []float32) (int, error)
-}
-
-// DoubleReader is an interface implemented by ValueReader instances which
-// expose the content of a column of double-precision float point values.
-type DoubleReader interface {
-	// Read double-precision floating point values into the buffer passed as
-	// argument.
-	//
-	// The method returns io.EOF when all values have been read.
-	ReadDoubles(values []float64) (int, error)
-}
-
-// DoubleWriter is an interface implemented by ValueWriter instances which
-// support writing columns of double-precision floating point values.
-type DoubleWriter interface {
-	// Write double-precision floating point values.
-	//
-	// The method returns the number of values written, and any error that
-	// occured while writing the values.
-	WriteDoubles(values []float64) (int, error)
-}
-
-// ByteArrayReader is an interface implemented by ValueReader instances which
-// expose the content of a column of variable length byte array values.
-type ByteArrayReader interface {
-	// Read values into the byte buffer passed as argument, returning the number
-	// of values written to the buffer (not the number of bytes). Values are
-	// written using the PLAIN encoding, each byte array prefixed with its
-	// length encoded as a 4 bytes little endian unsigned integer.
-	//
-	// The method returns io.EOF when all values have been read.
-	//
-	// If the buffer was not empty, but too small to hold at least one value,
-	// io.ErrShortBuffer is returned.
-	ReadByteArrays(values []byte) (int, error)
-}
-
-// ByteArrayWriter is an interface implemented by ValueWriter instances which
-// support writing columns of variable length byte array values.
-type ByteArrayWriter interface {
-	// Write variable length byte array values.
-	//
-	// The values passed as input must be laid out using the PLAIN encoding,
-	// with each byte array prefixed with the four bytes little endian unsigned
-	// integer length.
-	//
-	// The method returns the number of values written to the underlying column
-	// (not the number of bytes), or any error that occured while attempting to
-	// write the values.
-	WriteByteArrays(values []byte) (int, error)
-}
-
-// FixedLenByteArrayReader is an interface implemented by ValueReader instances
-// which expose the content of a column of fixed length byte array values.
-type FixedLenByteArrayReader interface {
-	// Read values into the byte buffer passed as argument, returning the number
-	// of values written to the buffer (not the number of bytes).
-	//
-	// The method returns io.EOF when all values have been read.
-	//
-	// If the buffer was not empty, but too small to hold at least one value,
-	// io.ErrShortBuffer is returned.
-	ReadFixedLenByteArrays(values []byte) (int, error)
-}
-
-// FixedLenByteArrayWriter is an interface implemented by ValueWriter instances
-// which support writing columns of fixed length byte array values.
-type FixedLenByteArrayWriter interface {
-	// Writes the fixed length byte array values.
-	//
-	// The size of the values is assumed to be the same as the expected size of
-	// items in the column. The method errors if the length of the input values
-	// is not a multiple of the expected item size.
-	WriteFixedLenByteArrays(values []byte) (int, error)
-}
-
 // ValueOf constructs a parquet value from a Go value v.
 //
 // The physical type of the value is assumed from the Go type of v using the
diff --git a/value_default.go b/value_default.go
new file mode 100644
index 0000000..624ff5e
--- /dev/null
+++ b/value_default.go
@@ -0,0 +1,175 @@
+//go:build !go1.18
+
+package parquet
+
+import "github.com/segmentio/parquet-go/deprecated"
+
+// BooleanReader is an interface implemented by ValueReader instances which
+// expose the content of a column of boolean values.
+type BooleanReader interface {
+	// Read boolean values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadBooleans(values []bool) (int, error)
+}
+
+// BooleanWriter is an interface implemented by ValueWriter instances which
+// support writing columns of boolean values.
+type BooleanWriter interface {
+	// Write boolean values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteBooleans(values []bool) (int, error)
+}
+
+// Int32Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int32 values.
+type Int32Reader interface {
+	// Read 32 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt32s(values []int32) (int, error)
+}
+
+// Int32Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 32 bits signed integer values.
+type Int32Writer interface {
+	// Write 32 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt32s(values []int32) (int, error)
+}
+
+// Int64Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int64 values.
+type Int64Reader interface {
+	// Read 64 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt64s(values []int64) (int, error)
+}
+
+// Int64Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 64 bits signed integer values.
+type Int64Writer interface {
+	// Write 64 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt64s(values []int64) (int, error)
+}
+
+// Int96Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int96 values.
+type Int96Reader interface {
+	// Read 96 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt96s(values []deprecated.Int96) (int, error)
+}
+
+// Int96Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 96 bits signed integer values.
+type Int96Writer interface {
+	// Write 96 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt96s(values []deprecated.Int96) (int, error)
+}
+
+// FloatReader is an interface implemented by ValueReader instances which expose
+// the content of a column of single-precision floating point values.
+type FloatReader interface {
+	// Read single-precision floating point values into the buffer passed as
+	// argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadFloats(values []float32) (int, error)
+}
+
+// FloatWriter is an interface implemented by ValueWriter instances which
+// support writing columns of single-precision floating point values.
+type FloatWriter interface {
+	// Write single-precision floating point values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteFloats(values []float32) (int, error)
+}
+
+// DoubleReader is an interface implemented by ValueReader instances which
+// expose the content of a column of double-precision float point values.
+type DoubleReader interface {
+	// Read double-precision floating point values into the buffer passed as
+	// argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadDoubles(values []float64) (int, error)
+}
+
+// DoubleWriter is an interface implemented by ValueWriter instances which
+// support writing columns of double-precision floating point values.
+type DoubleWriter interface {
+	// Write double-precision floating point values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteDoubles(values []float64) (int, error)
+}
+
+// ByteArrayReader is an interface implemented by ValueReader instances which
+// expose the content of a column of variable length byte array values.
+type ByteArrayReader interface {
+	// Read values into the byte buffer passed as argument, returning the number
+	// of values written to the buffer (not the number of bytes). Values are
+	// written using the PLAIN encoding, each byte array prefixed with its
+	// length encoded as a 4 bytes little endian unsigned integer.
+	//
+	// The method returns io.EOF when all values have been read.
+	//
+	// If the buffer was not empty, but too small to hold at least one value,
+	// io.ErrShortBuffer is returned.
+	ReadByteArrays(values []byte) (int, error)
+}
+
+// ByteArrayWriter is an interface implemented by ValueWriter instances which
+// support writing columns of variable length byte array values.
+type ByteArrayWriter interface {
+	// Write variable length byte array values.
+	//
+	// The values passed as input must be laid out using the PLAIN encoding,
+	// with each byte array prefixed with the four bytes little endian unsigned
+	// integer length.
+	//
+	// The method returns the number of values written to the underlying column
+	// (not the number of bytes), or any error that occured while attempting to
+	// write the values.
+	WriteByteArrays(values []byte) (int, error)
+}
+
+// FixedLenByteArrayReader is an interface implemented by ValueReader instances
+// which expose the content of a column of fixed length byte array values.
+type FixedLenByteArrayReader interface {
+	// Read values into the byte buffer passed as argument, returning the number
+	// of values written to the buffer (not the number of bytes).
+	//
+	// The method returns io.EOF when all values have been read.
+	//
+	// If the buffer was not empty, but too small to hold at least one value,
+	// io.ErrShortBuffer is returned.
+	ReadFixedLenByteArrays(values []byte) (int, error)
+}
+
+// FixedLenByteArrayWriter is an interface implemented by ValueWriter instances
+// which support writing columns of fixed length byte array values.
+type FixedLenByteArrayWriter interface {
+	// Writes the fixed length byte array values.
+	//
+	// The size of the values is assumed to be the same as the expected size of
+	// items in the column. The method errors if the length of the input values
+	// is not a multiple of the expected item size.
+	WriteFixedLenByteArrays(values []byte) (int, error)
+}
diff --git a/value_go18.go b/value_go18.go
new file mode 100644
index 0000000..9ad34c4
--- /dev/null
+++ b/value_go18.go
@@ -0,0 +1,35 @@
+//go:build go1.18
+
+package parquet
+
+import "github.com/segmentio/parquet-go/encoding/plain"
+
+// RequiredReader is a parameterized interface implemented by ValueReader
+// instances which exposes the content of a column as array of Go values of the
+// type parameter T.
+type RequiredReader[T plain.Type] interface {
+	// Read values into the data slice, returning the number of values read, or
+	// an error if less than len(data) values could be read, or io.EOF if the
+	// end of the sequence was reached.
+	//
+	// For columns of type BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY, T is byte and
+	// the data is PLAIN encoded.
+	//
+	// If the column is of type FIXED_LEN_BYTE_ARRAY, the data slice length must
+	// be a multiple of the column size.
+	ReadRequired(data []T) (int, error)
+}
+
+// RequiredWriter is a parameterized interface implemented by ValueWriter
+// instances which allows writing arrays of Go values of the type parameter T.
+type RequiredWriter[T plain.Type] interface {
+	// Write values from the data slice, returning the number of values written,
+	// or an error if less than len(data) values were written.
+	//
+	// For columns of type BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY, T is byte and
+	// the data is PLAIN encoded.
+	//
+	// If the column is of type FIXED_LEN_BYTE_ARRAY, the data slice length must
+	// be a multiple of the column size.
+	WriteRequired(data []T) (int, error)
+}
diff --git a/writer.go b/writer.go
index 1cea895..537089a 100644
--- a/writer.go
+++ b/writer.go
@@ -762,6 +762,12 @@ func (c *writerColumn) flushFilterPages() error {
 		if c.page.filter == nil {
 			c.page.filter = c.newBloomFilterEncoder(numValues)
 		}
+
+		// If there is a dictionary, we need to only write the dictionary.
+		if dict := c.dictionary; dict != nil {
+			return dict.WriteTo(c.page.filter)
+		}
+
 		for _, page := range c.filter {
 			if err := page.WriteTo(c.page.filter); err != nil {
 				return err


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 656e1fa..d25d6d8 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -8,6 +8,10 @@ jobs:
       matrix:
         go:
         - 1.17
+        - 1.18
+        tags:
+        - ''
+        - purego
 
     runs-on: ubuntu-latest
 
@@ -26,4 +30,4 @@ jobs:
       run: test -z $(find . -name '*.go' | xargs gofmt -l)
 
     - name: Run Tests
-      run: go test ./...
+      run: go test -race -tags=${{ matrix.tags }} ./...
diff --git a/column_test.go b/column_test.go
index f726f67..ca5f63c 100644
--- a/column_test.go
+++ b/column_test.go
@@ -1,9 +1,7 @@
 package parquet_test
 
 import (
-	"bytes"
 	"fmt"
-	"reflect"
 	"testing"
 	"testing/quick"
 
@@ -145,16 +143,14 @@ func checkColumnChunkColumnIndex(columnChunk parquet.ColumnChunk) error {
 	numPages := columnIndex.NumPages()
 	pagesRead := 0
 	err := forEachPage(columnChunk.Pages(), func(page parquet.Page) error {
-		min, max := page.Bounds()
-		pageMin := min.Bytes()
-		pageMax := max.Bytes()
+		pageMin, pageMax := page.Bounds()
 		indexMin := columnIndex.MinValue(pagesRead)
 		indexMax := columnIndex.MaxValue(pagesRead)
 
-		if !bytes.Equal(pageMin, indexMin) {
+		if !parquet.Equal(pageMin, indexMin) {
 			return fmt.Errorf("max page value mismatch: index=%x page=%x", indexMin, pageMin)
 		}
-		if !bytes.Equal(pageMax, indexMax) {
+		if !parquet.Equal(pageMax, indexMax) {
 			return fmt.Errorf("max page value mismatch: index=%x page=%x", indexMax, pageMax)
 		}
 
@@ -183,12 +179,12 @@ func checkColumnChunkColumnIndex(columnChunk parquet.ColumnChunk) error {
 
 		switch {
 		case columnIndex.IsAscending():
-			if columnType.Compare(min, max) > 0 {
-				return fmt.Errorf("column index is declared to be in ascending order but %v > %v", min, max)
+			if columnType.Compare(pageMin, pageMax) > 0 {
+				return fmt.Errorf("column index is declared to be in ascending order but %v > %v", pageMin, pageMax)
 			}
 		case columnIndex.IsDescending():
-			if columnType.Compare(min, max) < 0 {
-				return fmt.Errorf("column index is declared to be in desending order but %v < %v", min, max)
+			if columnType.Compare(pageMin, pageMax) < 0 {
+				return fmt.Errorf("column index is declared to be in desending order but %v < %v", pageMin, pageMax)
 			}
 		}
 
@@ -278,9 +274,57 @@ func checkFileColumnIndex(f *parquet.File) error {
 		if i >= len(columnIndexes) {
 			return fmt.Errorf("more column indexes were read when iterating over column chunks than when reading from the file (i=%d,n=%d)", i, len(columnIndexes))
 		}
-		if !reflect.DeepEqual(&columnIndexes[i], newColumnIndex(columnIndex)) {
-			return fmt.Errorf("column index at index %d mismatch:\nfile  = %#v\nchunk = %#v", i, &columnIndexes[i], columnIndex)
+
+		index1 := columnIndex
+		index2 := &fileColumnIndex{
+			kind:        col.Type().Kind(),
+			ColumnIndex: columnIndexes[i],
+		}
+
+		numPages1 := index1.NumPages()
+		numPages2 := index2.NumPages()
+		if numPages1 != numPages2 {
+			return fmt.Errorf("number of pages mismatch: got=%d want=%d", numPages1, numPages2)
 		}
+
+		for j := 0; j < numPages1; j++ {
+			nullCount1 := index1.NullCount(j)
+			nullCount2 := index2.NullCount(j)
+			if nullCount1 != nullCount2 {
+				return fmt.Errorf("null count of page %d/%d mismatch: got=%d want=%d", i, numPages1, nullCount1, nullCount2)
+			}
+
+			nullPage1 := index1.NullPage(j)
+			nullPage2 := index2.NullPage(j)
+			if nullPage1 != nullPage2 {
+				return fmt.Errorf("null page of page %d/%d mismatch: got=%t want=%t", i, numPages1, nullPage1, nullPage2)
+			}
+
+			minValue1 := index1.MinValue(j)
+			minValue2 := index2.MinValue(j)
+			if !parquet.Equal(minValue1, minValue2) {
+				return fmt.Errorf("min value of page %d/%d mismatch: got=%v want=%v", i, numPages1, minValue1, minValue2)
+			}
+
+			maxValue1 := index1.MaxValue(j)
+			maxValue2 := index2.MaxValue(j)
+			if !parquet.Equal(maxValue1, maxValue2) {
+				return fmt.Errorf("max value of page %d/%d mismatch: got=%v want=%v", i, numPages1, maxValue1, maxValue2)
+			}
+
+			isAscending1 := index1.IsAscending()
+			isAscending2 := index2.IsAscending()
+			if isAscending1 != isAscending2 {
+				return fmt.Errorf("ascending state of page %d/%d mismatch: got=%t want=%t", i, numPages1, isAscending1, isAscending2)
+			}
+
+			isDescending1 := index1.IsDescending()
+			isDescending2 := index2.IsDescending()
+			if isDescending1 != isDescending2 {
+				return fmt.Errorf("descending state of page %d/%d mismatch: got=%t want=%t", i, numPages1, isDescending1, isDescending2)
+			}
+		}
+
 		i++
 		return nil
 	})
@@ -297,53 +341,59 @@ func checkFileOffsetIndex(f *parquet.File) error {
 		if i >= len(offsetIndexes) {
 			return fmt.Errorf("more offset indexes were read when iterating over column chunks than when reading from the file (i=%d,n=%d)", i, len(offsetIndexes))
 		}
-		if !reflect.DeepEqual(&offsetIndexes[i], newOffsetIndex(offsetIndex)) {
-			return fmt.Errorf("offset index at index %d mismatch:\nfile  = %#v\nchunk = %#v", i, &offsetIndexes[i], offsetIndex)
+
+		index1 := offsetIndex
+		index2 := (*fileOffsetIndex)(&offsetIndexes[i])
+
+		numPages1 := index1.NumPages()
+		numPages2 := index2.NumPages()
+		if numPages1 != numPages2 {
+			return fmt.Errorf("number of pages mismatch: got=%d want=%d", numPages1, numPages2)
+		}
+
+		for j := 0; j < numPages1; j++ {
+			offset1 := index1.Offset(j)
+			offset2 := index2.Offset(j)
+			if offset1 != offset2 {
+				return fmt.Errorf("offsets of page %d/%d mismatch: got=%d want=%d", i, numPages1, offset1, offset2)
+			}
+
+			compressedPageSize1 := index1.CompressedPageSize(j)
+			compressedPageSize2 := index2.CompressedPageSize(j)
+			if compressedPageSize1 != compressedPageSize2 {
+				return fmt.Errorf("compressed page size of page %d/%d mismatch: got=%d want=%d", i, numPages1, compressedPageSize1, compressedPageSize2)
+			}
+
+			firstRowIndex1 := index1.FirstRowIndex(j)
+			firstRowIndex2 := index2.FirstRowIndex(j)
+			if firstRowIndex1 != firstRowIndex2 {
+				return fmt.Errorf("first row index of page %d/%d mismatch: got=%d want=%d", i, numPages1, firstRowIndex1, firstRowIndex2)
+			}
 		}
+
 		i++
 		return nil
 	})
 }
 
-func newColumnIndex(columnIndex parquet.ColumnIndex) *format.ColumnIndex {
-	numPages := columnIndex.NumPages()
-
-	index := &format.ColumnIndex{
-		NullPages:  make([]bool, numPages),
-		MinValues:  make([][]byte, numPages),
-		MaxValues:  make([][]byte, numPages),
-		NullCounts: make([]int64, numPages),
-	}
-
-	for i := 0; i < numPages; i++ {
-		index.NullPages[i] = columnIndex.NullPage(i)
-		index.MinValues[i] = columnIndex.MinValue(i)
-		index.MaxValues[i] = columnIndex.MaxValue(i)
-		index.NullCounts[i] = columnIndex.NullCount(i)
-	}
-
-	switch {
-	case columnIndex.IsAscending():
-		index.BoundaryOrder = format.Ascending
-	case columnIndex.IsDescending():
-		index.BoundaryOrder = format.Descending
-	}
-
-	return index
+type fileColumnIndex struct {
+	kind parquet.Kind
+	format.ColumnIndex
 }
 
-func newOffsetIndex(offsetIndex parquet.OffsetIndex) *format.OffsetIndex {
-	index := &format.OffsetIndex{
-		PageLocations: make([]format.PageLocation, offsetIndex.NumPages()),
-	}
+func (i *fileColumnIndex) NumPages() int                { return len(i.NullPages) }
+func (i *fileColumnIndex) NullCount(j int) int64        { return i.NullCounts[j] }
+func (i *fileColumnIndex) NullPage(j int) bool          { return i.NullPages[j] }
+func (i *fileColumnIndex) MinValue(j int) parquet.Value { return i.kind.Value(i.MinValues[j]) }
+func (i *fileColumnIndex) MaxValue(j int) parquet.Value { return i.kind.Value(i.MaxValues[j]) }
+func (i *fileColumnIndex) IsAscending() bool            { return i.BoundaryOrder == format.Ascending }
+func (i *fileColumnIndex) IsDescending() bool           { return i.BoundaryOrder == format.Descending }
 
-	for i := range index.PageLocations {
-		index.PageLocations[i] = format.PageLocation{
-			Offset:             offsetIndex.Offset(i),
-			CompressedPageSize: int32(offsetIndex.CompressedPageSize(i)),
-			FirstRowIndex:      offsetIndex.FirstRowIndex(i),
-		}
-	}
+type fileOffsetIndex format.OffsetIndex
 
-	return index
+func (i *fileOffsetIndex) NumPages() int      { return len(i.PageLocations) }
+func (i *fileOffsetIndex) Offset(j int) int64 { return i.PageLocations[j].Offset }
+func (i *fileOffsetIndex) CompressedPageSize(j int) int64 {
+	return int64(i.PageLocations[j].CompressedPageSize)
 }
+func (i *fileOffsetIndex) FirstRowIndex(j int) int64 { return i.PageLocations[j].FirstRowIndex }
diff --git a/page_go18_test.go b/page_go18_test.go
new file mode 100644
index 0000000..a06e62a
--- /dev/null
+++ b/page_go18_test.go
@@ -0,0 +1,342 @@
+//go:build go1.18
+
+package parquet_test
+
+import (
+	"bytes"
+	"io"
+	"math/rand"
+	"reflect"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding/plain"
+	"github.com/segmentio/parquet-go/internal/cast"
+)
+
+func TestPage(t *testing.T) {
+	t.Run("BOOLEAN", testPageOf[bool])
+	t.Run("INT32", testPageOf[int32])
+	t.Run("INT64", testPageOf[int64])
+	t.Run("INT96", testPageOf[deprecated.Int96])
+	t.Run("FLOAT", testPageOf[float32])
+	t.Run("DOUBLE", testPageOf[float64])
+	t.Run("BYTE_ARRAY", testPageByteArray)
+	t.Run("FIXED_LEN_BYTE_ARRAY", testPageFixedLenByteArray)
+}
+
+func randValue[T plain.Type](r *rand.Rand) (v T) {
+	switch (any)(v).(type) {
+	case bool:
+		return (any)(r.Float64() < 0.5).(T)
+	case int32:
+		return (any)(r.Int31()).(T)
+	case int64:
+		return (any)(r.Int63()).(T)
+	case deprecated.Int96:
+		return (any)(deprecated.Int96{
+			0: r.Uint32(),
+			1: r.Uint32(),
+			2: r.Uint32(),
+		}).(T)
+	case float32:
+		return (any)(r.Float32()).(T)
+	case float64:
+		return (any)(r.Float64()).(T)
+	default:
+		panic("NOT IMPLEMENTED")
+	}
+}
+
+func testPageOf[T plain.Type](t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value T }{})
+	r := rand.New(rand.NewSource(0))
+
+	t.Run("io", func(t *testing.T) {
+		testBufferPage(t, schema, pageTest[T]{
+			write: func(w parquet.ValueWriter) ([]T, error) {
+				values := []T{
+					0: randValue[T](r),
+					1: randValue[T](r),
+				}
+				n, err := w.(io.Writer).Write(cast.SliceToBytes(values))
+				return values[:n], err
+			},
+			read: func(r parquet.ValueReader) ([]T, error) {
+				values := make([]T, 2)
+				n, err := r.(io.Reader).Read(cast.SliceToBytes(values))
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest[T]{
+			write: func(w parquet.ValueWriter) ([]T, error) {
+				values := []T{
+					0: randValue[T](r),
+					1: randValue[T](r),
+				}
+				n, err := w.(parquet.RequiredWriter[T]).WriteRequired(values)
+				return values[:n], err
+			},
+			read: func(r parquet.ValueReader) ([]T, error) {
+				values := make([]T, 2)
+				n, err := r.(parquet.RequiredReader[T]).ReadRequired(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageByteArray(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value []byte }{})
+
+	t.Run("io", func(t *testing.T) {
+		testBufferPage(t, schema, pageTest[byte]{
+			write: func(w parquet.ValueWriter) ([]byte, error) {
+				values := []byte{}
+				values = plain.AppendByteArray(values, []byte("A"))
+				values = plain.AppendByteArray(values, []byte("B"))
+				values = plain.AppendByteArray(values, []byte("C"))
+				n, err := w.(io.Writer).Write(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) ([]byte, error) {
+				values := make([]byte, 3+3*plain.ByteArrayLengthSize)
+				n, err := r.(io.Reader).Read(values)
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest[byte]{
+			write: func(w parquet.ValueWriter) ([]byte, error) {
+				values := []byte{}
+				values = plain.AppendByteArray(values, []byte("A"))
+				values = plain.AppendByteArray(values, []byte("B"))
+				values = plain.AppendByteArray(values, []byte("C"))
+				_, err := w.(parquet.RequiredWriter[byte]).WriteRequired(values)
+				return values, err
+			},
+
+			read: func(r parquet.ValueReader) ([]byte, error) {
+				values := make([]byte, 3+3*plain.ByteArrayLengthSize)
+				n, err := r.(parquet.RequiredReader[byte]).ReadRequired(values)
+				return values[:n+n*plain.ByteArrayLengthSize], err
+			},
+		})
+	})
+}
+
+func testPageFixedLenByteArray(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value [3]byte }{})
+
+	t.Run("io", func(t *testing.T) {
+		testBufferPage(t, schema, pageTest[byte]{
+			write: func(w parquet.ValueWriter) ([]byte, error) {
+				values := []byte("123456789")
+				n, err := w.(io.Writer).Write(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) ([]byte, error) {
+				values := make([]byte, 3*3)
+				n, err := r.(io.Reader).Read(values)
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest[byte]{
+			write: func(w parquet.ValueWriter) ([]byte, error) {
+				values := []byte("123456789")
+				n, err := w.(parquet.RequiredWriter[byte]).WriteRequired(values)
+				return values[:3*n], err
+			},
+
+			read: func(r parquet.ValueReader) ([]byte, error) {
+				values := make([]byte, 3*3)
+				n, err := r.(parquet.RequiredReader[byte]).ReadRequired(values)
+				return values[:3*n], err
+			},
+		})
+	})
+}
+
+type pageTest[T any] struct {
+	write func(parquet.ValueWriter) ([]T, error)
+	read  func(parquet.ValueReader) ([]T, error)
+}
+
+func testPage[T any](t *testing.T, schema *parquet.Schema, test pageTest[T]) {
+	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 any](t *testing.T, schema *parquet.Schema, test pageTest[T]) {
+	buffer := parquet.NewBuffer(schema)
+	column := buffer.Column(0).(parquet.ColumnBuffer)
+
+	w, err := test.write(column)
+	if err != nil {
+		t.Fatal("writing page values:", err)
+	}
+
+	r, err := test.read(column.Page().Values())
+	if err != io.EOF {
+		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)
+	}
+}
+
+func testFilePage[T any](t *testing.T, schema *parquet.Schema, test pageTest[T]) {
+	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 %v 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))
+	}
+}
+
+func TestOptionalPagePreserveIndex(t *testing.T) {
+	schema := parquet.SchemaOf(&testStruct{})
+	buffer := parquet.NewBuffer(schema)
+
+	if err := buffer.WriteRow(schema.Deconstruct(nil, &testStruct{Value: nil})); err != nil {
+		t.Fatal("writing row:", err)
+	}
+
+	row, err := buffer.Rows().ReadRow(nil)
+	if err != nil {
+		t.Fatal("reading rows:", err)
+	}
+
+	if row[0].Column() != 0 {
+		t.Errorf("wrong index: got=%d want=%d", row[0].Column(), 0)
+	}
+}
+
+func TestRepeatedPageTrailingNulls(t *testing.T) {
+	type testStruct struct {
+		A []string `parquet:"a"`
+	}
+
+	s := parquet.SchemaOf(&testStruct{})
+
+	records := []*testStruct{
+		{A: nil},
+		{A: []string{"test"}},
+		{A: nil},
+	}
+
+	buf := parquet.NewBuffer(s)
+	for _, rec := range records {
+		row := s.Deconstruct(nil, rec)
+		err := buf.WriteRow(row)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	resultRows := []parquet.Row{}
+	reader := buf.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(records) {
+		t.Errorf("wrong number of rows read: got=%d want=%d", len(resultRows), len(records))
+	}
+}
diff --git a/page_test.go b/page_test.go
index 284a259..ee256a9 100644
--- a/page_test.go
+++ b/page_test.go
@@ -1,3 +1,5 @@
+//go:build !go1.18
+
 package parquet_test
 
 import (
@@ -449,3 +451,43 @@ func TestOptionalPagePreserveIndex(t *testing.T) {
 		t.Errorf("wrong index: got=%d want=%d", row[0].Column(), 0)
 	}
 }
+
+func TestRepeatedPageTrailingNulls(t *testing.T) {
+	type testStruct struct {
+		A []string `parquet:"a"`
+	}
+
+	s := parquet.SchemaOf(&testStruct{})
+
+	records := []*testStruct{
+		{A: nil},
+		{A: []string{"test"}},
+		{A: nil},
+	}
+
+	buf := parquet.NewBuffer(s)
+	for _, rec := range records {
+		row := s.Deconstruct(nil, rec)
+		err := buf.WriteRow(row)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	resultRows := []parquet.Row{}
+	reader := buf.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(records) {
+		t.Errorf("wrong number of rows read: got=%d want=%d", len(resultRows), len(records))
+	}
+}
diff --git a/sort_test.go b/sort_test.go
new file mode 100644
index 0000000..3cb946a
--- /dev/null
+++ b/sort_test.go
@@ -0,0 +1,82 @@
+package parquet_test
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/segmentio/parquet-go"
+)
+
+func TestSortFunc(t *testing.T) {
+	sortFunc := parquet.SortFuncOf(parquet.String().Type(),
+		parquet.SortMaxDefinitionLevel(1),
+		parquet.SortDescending(true),
+		parquet.SortNullsFirst(true),
+	)
+
+	values := [][]parquet.Value{
+		{parquet.ValueOf("A")},
+		{parquet.ValueOf(nil)},
+		{parquet.ValueOf(nil)},
+		{parquet.ValueOf("C")},
+		{parquet.ValueOf("B")},
+		{parquet.ValueOf(nil)},
+	}
+
+	expect := [][]parquet.Value{
+		{parquet.ValueOf(nil)},
+		{parquet.ValueOf(nil)},
+		{parquet.ValueOf(nil)},
+		{parquet.ValueOf("C")},
+		{parquet.ValueOf("B")},
+		{parquet.ValueOf("A")},
+	}
+
+	sort.Slice(values, func(i, j int) bool {
+		return sortFunc(values[i], values[j]) < 0
+	})
+
+	for i := range values {
+		if !parquet.Equal(values[i][0], expect[i][0]) {
+			t.Errorf("value at index %d mismatch: got=%+v want=%+v\n%+v\n%+v", i, expect[i], values[i], expect, values)
+			break
+		}
+	}
+}
+
+func TestRepeatedUUIDSortFunc(t *testing.T) {
+	type testStruct struct {
+		List []uuid.UUID `parquet:"list"`
+	}
+
+	s := parquet.SchemaOf(&testStruct{})
+
+	a := s.Deconstruct(nil, &testStruct{
+		List: []uuid.UUID{
+			uuid.MustParse("00000000-0000-0000-0000-000000000001"),
+			uuid.MustParse("00000000-0000-0000-0000-000000000002"),
+		},
+	})
+
+	b := s.Deconstruct(nil, &testStruct{
+		List: []uuid.UUID{
+			uuid.MustParse("00000000-0000-0000-0000-000000000001"),
+			uuid.MustParse("00000000-0000-0000-0000-000000000002"),
+			uuid.MustParse("00000000-0000-0000-0000-000000000003"),
+		},
+	})
+
+	// a and b are equal up until the third element, then a ends so a < b.
+	f := parquet.SortFuncOf(
+		s.ChildByName("list").Type(),
+		parquet.SortDescending(false),
+		parquet.SortNullsFirst(true),
+		parquet.SortMaxDefinitionLevel(1),
+		parquet.SortMaxRepetitionLevel(1),
+	)
+	cmp := f(a, b)
+	if cmp >= 0 {
+		t.Fatal("expected a < b, got compare value", cmp)
+	}
+}
diff --git a/writer_test.go b/writer_test.go
index b4ab7a2..470e4c4 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -527,3 +527,41 @@ func TestWriterGenerateBloomFilters(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+func TestBloomFilterForDict(t *testing.T) {
+	type testStruct struct {
+		A string `parquet:"a,dict"`
+	}
+
+	schema := parquet.SchemaOf(&testStruct{})
+
+	b := bytes.NewBuffer(nil)
+	w := parquet.NewWriter(
+		b,
+		schema,
+		parquet.BloomFilters(parquet.SplitBlockFilter("a")),
+	)
+
+	err := w.Write(&testStruct{A: "test"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = w.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	f, err := parquet.OpenFile(bytes.NewReader(b.Bytes()), int64(b.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ok, err := f.RowGroup(0).Column(0).BloomFilter().Check(parquet.ValueOf("test"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !ok {
+		t.Error("bloom filter should have contained 'test'")
+	}
+}