Repository: segmentio/parquet-go
Pull request #103 (2022-03-17 01:24:17)

refactor primitive parquet types to leverage Go 1.18 generics
Follow up to #98, this PR modifies more of the internal types to use Go 1.18 generics in order to reduce the amount of code that we have to maintain.


Patch:
diff --git a/class_go18.go b/class_go18.go
index 052731b..1b48b4a 100644
--- a/class_go18.go
+++ b/class_go18.go
@@ -42,6 +42,8 @@ func less[T comparable](a, b T) bool {
 
 type class[T primitive] struct {
 	name      string
+	bits      int8
+	kind      Kind
 	makeValue func(T) Value
 	value     func(Value) T
 	plain     func(T) []byte
@@ -57,6 +59,8 @@ type class[T primitive] struct {
 
 var boolClass = class[bool]{
 	name:      "BOOLEAN",
+	bits:      1,
+	kind:      Boolean,
 	makeValue: makeValueBoolean,
 	value:     Value.Boolean,
 	plain:     plain.Boolean,
@@ -72,6 +76,8 @@ var boolClass = class[bool]{
 
 var int32Class = class[int32]{
 	name:      "INT32",
+	bits:      32,
+	kind:      Int32,
 	makeValue: makeValueInt32,
 	value:     Value.Int32,
 	plain:     plain.Int32,
@@ -87,6 +93,8 @@ var int32Class = class[int32]{
 
 var int64Class = class[int64]{
 	name:      "INT64",
+	bits:      64,
+	kind:      Int64,
 	makeValue: makeValueInt64,
 	value:     Value.Int64,
 	plain:     plain.Int64,
@@ -102,6 +110,8 @@ var int64Class = class[int64]{
 
 var int96Class = class[deprecated.Int96]{
 	name:      "INT96",
+	bits:      96,
+	kind:      Int96,
 	makeValue: makeValueInt96,
 	value:     Value.Int96,
 	plain:     plain.Int96,
@@ -117,6 +127,8 @@ var int96Class = class[deprecated.Int96]{
 
 var float32Class = class[float32]{
 	name:      "FLOAT",
+	bits:      32,
+	kind:      Float,
 	makeValue: makeValueFloat,
 	value:     Value.Float,
 	plain:     plain.Float,
@@ -132,6 +144,8 @@ var float32Class = class[float32]{
 
 var float64Class = class[float64]{
 	name:      "DOUBLE",
+	bits:      64,
+	kind:      Double,
 	makeValue: makeValueDouble,
 	value:     Value.Double,
 	plain:     plain.Double,
@@ -147,6 +161,8 @@ var float64Class = class[float64]{
 
 var uint32Class = class[uint32]{
 	name:      "INT32",
+	bits:      32,
+	kind:      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)) },
@@ -166,6 +182,8 @@ var uint32Class = class[uint32]{
 
 var uint64Class = class[uint64]{
 	name:      "INT64",
+	bits:      64,
+	kind:      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)) },
diff --git a/column_buffer.go b/column_buffer.go
index b2bc179..9b7f16f 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -182,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
 
@@ -256,26 +263,33 @@ func (col *optionalColumnBuffer) WriteValues(values []Value) (n int, err error)
 	rowIndex := int32(col.base.Len())
 
 	for n < len(values) {
+		// Collect index range of contiguous null values, from i to n. If this
+		// for loop exhausts the values, all remaining if statements and for
+		// loops will be no-ops and the loop will terminate.
 		i := n
 		for n < len(values) && values[n].definitionLevel != col.maxDefinitionLevel {
 			n++
 		}
 
+		// Write the contiguous null values up until the first non-null value
+		// obtained in the for loop above.
 		for _, v := range values[i:n] {
-			col.rows = append(col.rows, rowIndex)
+			col.rows = append(col.rows, -1)
 			col.definitionLevels = append(col.definitionLevels, v.definitionLevel)
-			rowIndex++
 		}
 
+		// Collect index range of contiguous non-null values, from i to n.
 		i = n
 		for n < len(values) && values[n].definitionLevel == col.maxDefinitionLevel {
 			n++
 		}
 
+		// As long as i < n we have non-null values still to write. It is
+		// possible that we just exhausted the input values in which case i == n
+		// and the outer for loop will terminate.
 		if i < n {
 			count, err := col.base.WriteValues(values[i:n])
 			col.definitionLevels = appendLevel(col.definitionLevels, col.maxDefinitionLevel, count)
-			n += count
 
 			for count > 0 {
 				col.rows = append(col.rows, rowIndex)
@@ -432,7 +446,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)
 		}
@@ -642,7 +656,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 {
@@ -690,11 +724,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 }
@@ -804,11 +838,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 }
diff --git a/column_buffer_default.go b/column_buffer_default.go
index 05031df..1d390bd 100644
--- a/column_buffer_default.go
+++ b/column_buffer_default.go
@@ -49,9 +49,13 @@ func (col *booleanColumnBuffer) Clone() ColumnBuffer {
 
 func (col *booleanColumnBuffer) Type() Type { return col.typ }
 
-func (col *booleanColumnBuffer) ColumnIndex() ColumnIndex { return booleanPageIndex{&col.booleanPage} }
+func (col *booleanColumnBuffer) ColumnIndex() ColumnIndex {
+	return booleanColumnIndex{&col.booleanPage}
+}
 
-func (col *booleanColumnBuffer) OffsetIndex() OffsetIndex { return booleanPageIndex{&col.booleanPage} }
+func (col *booleanColumnBuffer) OffsetIndex() OffsetIndex {
+	return booleanOffsetIndex{&col.booleanPage}
+}
 
 func (col *booleanColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -142,9 +146,9 @@ func (col *int32ColumnBuffer) Clone() ColumnBuffer {
 
 func (col *int32ColumnBuffer) Type() Type { return col.typ }
 
-func (col *int32ColumnBuffer) ColumnIndex() ColumnIndex { return int32PageIndex{&col.int32Page} }
+func (col *int32ColumnBuffer) ColumnIndex() ColumnIndex { return int32ColumnIndex{&col.int32Page} }
 
-func (col *int32ColumnBuffer) OffsetIndex() OffsetIndex { return int32PageIndex{&col.int32Page} }
+func (col *int32ColumnBuffer) OffsetIndex() OffsetIndex { return int32OffsetIndex{&col.int32Page} }
 
 func (col *int32ColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -237,9 +241,9 @@ func (col *int64ColumnBuffer) Clone() ColumnBuffer {
 
 func (col *int64ColumnBuffer) Type() Type { return col.typ }
 
-func (col *int64ColumnBuffer) ColumnIndex() ColumnIndex { return int64PageIndex{&col.int64Page} }
+func (col *int64ColumnBuffer) ColumnIndex() ColumnIndex { return int64ColumnIndex{&col.int64Page} }
 
-func (col *int64ColumnBuffer) OffsetIndex() OffsetIndex { return int64PageIndex{&col.int64Page} }
+func (col *int64ColumnBuffer) OffsetIndex() OffsetIndex { return int64OffsetIndex{&col.int64Page} }
 
 func (col *int64ColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -332,9 +336,9 @@ func (col *int96ColumnBuffer) Clone() ColumnBuffer {
 
 func (col *int96ColumnBuffer) Type() Type { return col.typ }
 
-func (col *int96ColumnBuffer) ColumnIndex() ColumnIndex { return int96PageIndex{&col.int96Page} }
+func (col *int96ColumnBuffer) ColumnIndex() ColumnIndex { return int96ColumnIndex{&col.int96Page} }
 
-func (col *int96ColumnBuffer) OffsetIndex() OffsetIndex { return int96PageIndex{&col.int96Page} }
+func (col *int96ColumnBuffer) OffsetIndex() OffsetIndex { return int96OffsetIndex{&col.int96Page} }
 
 func (col *int96ColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -427,9 +431,9 @@ func (col *floatColumnBuffer) Clone() ColumnBuffer {
 
 func (col *floatColumnBuffer) Type() Type { return col.typ }
 
-func (col *floatColumnBuffer) ColumnIndex() ColumnIndex { return floatPageIndex{&col.floatPage} }
+func (col *floatColumnBuffer) ColumnIndex() ColumnIndex { return floatColumnIndex{&col.floatPage} }
 
-func (col *floatColumnBuffer) OffsetIndex() OffsetIndex { return floatPageIndex{&col.floatPage} }
+func (col *floatColumnBuffer) OffsetIndex() OffsetIndex { return floatOffsetIndex{&col.floatPage} }
 
 func (col *floatColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -522,9 +526,9 @@ func (col *doubleColumnBuffer) Clone() ColumnBuffer {
 
 func (col *doubleColumnBuffer) Type() Type { return col.typ }
 
-func (col *doubleColumnBuffer) ColumnIndex() ColumnIndex { return doublePageIndex{&col.doublePage} }
+func (col *doubleColumnBuffer) ColumnIndex() ColumnIndex { return doubleColumnIndex{&col.doublePage} }
 
-func (col *doubleColumnBuffer) OffsetIndex() OffsetIndex { return doublePageIndex{&col.doublePage} }
+func (col *doubleColumnBuffer) OffsetIndex() OffsetIndex { return doubleOffsetIndex{&col.doublePage} }
 
 func (col *doubleColumnBuffer) BloomFilter() BloomFilter { return nil }
 
@@ -596,9 +600,9 @@ func newUint32ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint32Co
 	return uint32ColumnBuffer{newInt32ColumnBuffer(typ, columnIndex, bufferSize)}
 }
 
-func (col uint32ColumnBuffer) ColumnIndex() ColumnIndex { return uint32PageIndex{col.page()} }
+func (col uint32ColumnBuffer) ColumnIndex() ColumnIndex { return uint32ColumnIndex{col.page()} }
 
-func (col uint32ColumnBuffer) OffsetIndex() OffsetIndex { return uint32PageIndex{col.page()} }
+func (col uint32ColumnBuffer) OffsetIndex() OffsetIndex { return uint32OffsetIndex{col.page()} }
 
 func (col uint32ColumnBuffer) page() uint32Page { return uint32Page{&col.int32Page} }
 
@@ -620,9 +624,9 @@ func newUint64ColumnBuffer(typ Type, columnIndex int16, bufferSize int) uint64Co
 	return uint64ColumnBuffer{newInt64ColumnBuffer(typ, columnIndex, bufferSize)}
 }
 
-func (col uint64ColumnBuffer) ColumnIndex() ColumnIndex { return uint64PageIndex{col.page()} }
+func (col uint64ColumnBuffer) ColumnIndex() ColumnIndex { return uint64ColumnIndex{col.page()} }
 
-func (col uint64ColumnBuffer) OffsetIndex() OffsetIndex { return uint64PageIndex{col.page()} }
+func (col uint64ColumnBuffer) OffsetIndex() OffsetIndex { return uint64OffsetIndex{col.page()} }
 
 func (col uint64ColumnBuffer) page() uint64Page { return uint64Page{&col.int64Page} }
 
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index 8dfc8af..d61f971 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/internal/cast"
 )
 
@@ -15,30 +14,6 @@ type columnBuffer[T primitive] struct {
 	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)
 }
@@ -71,9 +46,9 @@ func (col *columnBuffer[T]) Clone() ColumnBuffer {
 
 func (col *columnBuffer[T]) Type() Type { return col.typ }
 
-func (col *columnBuffer[T]) ColumnIndex() ColumnIndex { return pageIndex[T]{&col.page} }
+func (col *columnBuffer[T]) ColumnIndex() ColumnIndex { return columnIndex[T]{&col.page} }
 
-func (col *columnBuffer[T]) OffsetIndex() OffsetIndex { return pageIndex[T]{&col.page} }
+func (col *columnBuffer[T]) OffsetIndex() OffsetIndex { return offsetIndex[T]{&col.page} }
 
 func (col *columnBuffer[T]) BloomFilter() BloomFilter { return nil }
 
diff --git a/column_index.go b/column_index.go
index d1c4867..d23387b 100644
--- a/column_index.go
+++ b/column_index.go
@@ -20,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
@@ -34,35 +34,56 @@ type ColumnIndex interface {
 	IsDescending() bool
 }
 
-type columnIndex format.ColumnIndex
-
-func (i *columnIndex) NumPages() int         { return len(i.NullPages) }
-func (i *columnIndex) NullCount(j int) int64 { return i.NullCounts[j] }
-func (i *columnIndex) NullPage(j int) bool   { return i.NullPages[j] }
-func (i *columnIndex) MinValue(j int) []byte { return i.MinValues[j] }
-func (i *columnIndex) MaxValue(j int) []byte { return i.MaxValues[j] }
-func (i *columnIndex) IsAscending() bool     { return i.BoundaryOrder == format.Ascending }
-func (i *columnIndex) IsDescending() bool    { return i.BoundaryOrder == format.Descending }
-
-type byteArrayPageIndex struct{ page *byteArrayPage }
-
-func (i byteArrayPageIndex) NumPages() int       { return 1 }
-func (i byteArrayPageIndex) NullCount(int) int64 { return 0 }
-func (i byteArrayPageIndex) NullPage(int) bool   { return false }
-func (i byteArrayPageIndex) MinValue(int) []byte { return copyBytes(i.page.min()) }
-func (i byteArrayPageIndex) MaxValue(int) []byte { return copyBytes(i.page.max()) }
-func (i byteArrayPageIndex) IsAscending() bool   { return bytes.Compare(i.page.bounds()) < 0 }
-func (i byteArrayPageIndex) IsDescending() bool  { return bytes.Compare(i.page.bounds()) > 0 }
-
-type fixedLenByteArrayPageIndex struct{ page *fixedLenByteArrayPage }
-
-func (i fixedLenByteArrayPageIndex) NumPages() int       { return 1 }
-func (i fixedLenByteArrayPageIndex) NullCount(int) int64 { return 0 }
-func (i fixedLenByteArrayPageIndex) NullPage(int) bool   { return false }
-func (i fixedLenByteArrayPageIndex) MinValue(int) []byte { return copyBytes(i.page.min()) }
-func (i fixedLenByteArrayPageIndex) MaxValue(int) []byte { return copyBytes(i.page.max()) }
-func (i fixedLenByteArrayPageIndex) IsAscending() bool   { return bytes.Compare(i.page.bounds()) < 0 }
-func (i fixedLenByteArrayPageIndex) IsDescending() bool  { return bytes.Compare(i.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 (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 byteArrayColumnIndex struct{ page *byteArrayPage }
+
+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 fixedLenByteArrayColumnIndex struct{ page *fixedLenByteArrayPage }
+
+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.
diff --git a/column_index_default.go b/column_index_default.go
index 9ce976c..d9bace2 100644
--- a/column_index_default.go
+++ b/column_index_default.go
@@ -4,90 +4,89 @@ package parquet
 
 import (
 	"github.com/segmentio/parquet-go/deprecated"
-	"github.com/segmentio/parquet-go/encoding/plain"
 	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
-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 uint32PageIndex struct{ page uint32Page }
-
-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 }
-
-type uint64PageIndex struct{ page uint64Page }
-
-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 }
+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
diff --git a/column_index_go18.go b/column_index_go18.go
index b7e326c..9f593b1 100644
--- a/column_index_go18.go
+++ b/column_index_go18.go
@@ -3,28 +3,19 @@
 package parquet
 
 import (
-	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/cast"
 )
 
-type pageIndex[T primitive] struct{ page *page[T] }
+type columnIndex[T primitive] struct{ page *page[T] }
 
-// ColumnIndex
-
-func (i pageIndex[T]) NumPages() int       { return 1 }
-func (i pageIndex[T]) NullCount(int) int64 { return 0 }
-func (i pageIndex[T]) NullPage(int) bool   { return false }
-func (i pageIndex[T]) MinValue(int) []byte { return i.page.class.plain(i.page.min()) }
-func (i pageIndex[T]) MaxValue(int) []byte { return i.page.class.plain(i.page.max()) }
-func (i pageIndex[T]) IsAscending() bool   { return i.page.class.compare(i.page.bounds()) < 0 }
-func (i pageIndex[T]) IsDescending() bool  { return i.page.class.compare(i.page.bounds()) > 0 }
-
-// OffsetIndex
-
-func (i pageIndex[T]) Offset(int) int64             { return 0 }
-func (i pageIndex[T]) CompressedPageSize(int) int64 { return i.page.Size() }
-func (i pageIndex[T]) FirstRowIndex(int) int64      { return 0 }
+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]
@@ -34,6 +25,10 @@ type columnIndexer[T primitive] struct {
 	maxValues  []T
 }
 
+func newColumnIndexer[T primitive](class *class[T]) *columnIndexer[T] {
+	return &columnIndexer[T]{class: class}
+}
+
 func (i *columnIndexer[T]) Reset() {
 	i.nullPages = i.nullPages[:0]
 	i.nullCounts = i.nullCounts[:0]
@@ -61,35 +56,3 @@ func (i *columnIndexer[T]) ColumnIndex() format.ColumnIndex {
 		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_go18.go b/column_reader_go18.go
index 35a94ef..162f994 100644
--- a/column_reader_go18.go
+++ b/column_reader_go18.go
@@ -5,7 +5,6 @@ package parquet
 import (
 	"io"
 
-	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 )
 
@@ -19,38 +18,6 @@ type columnReader[T primitive] struct {
 	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,
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..08c5faf 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -1,10 +1,10 @@
 package parquet
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 
-	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
@@ -28,11 +28,14 @@ type Dictionary interface {
 	Len() int
 
 	// Returns the dictionary value at the given index.
-	Index(int) Value
+	Index(index int32) Value
 
-	// Inserts a value to the dictionary, returning the index at which it was
-	// recorded.
-	Insert(Value) int
+	// Inserts values from the second slice to the dictionary and writes the
+	// indexes at which each value was inserted to the first slice.
+	//
+	// The method panics if the length of the indexes slice is smaller than the
+	// length of the values slice.
+	Insert(indexes []int32, values []Value)
 
 	// Given an array of dictionary indexes, lookup the values into the array
 	// of values passed as second argument.
@@ -41,6 +44,9 @@ type Dictionary interface {
 	// is negative or greater than the highest index in the dictionary.
 	Lookup(indexes []int32, values []Value)
 
+	// Returns the min and max values found in the given indexes.
+	Bounds(indexed []int32) (min, max Value)
+
 	// Reads the dictionary from the decoder passed as argument.
 	//
 	// The dictionary is cleared prior to loading the values so that its final
@@ -59,463 +65,6 @@ func dictCap(bufferSize, valueItemSize int) int {
 	return atLeastOne(bufferSize / (valueItemSize + indexItemSize))
 }
 
-// The boolean dictionary always contains two values for true and false.
-type booleanDictionary struct {
-	typ    Type
-	values [2]bool
-}
-
-func newBooleanDictionary(typ Type) *booleanDictionary {
-	return &booleanDictionary{
-		typ:    typ,
-		values: [2]bool{false, true},
-	}
-}
-
-func (d *booleanDictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *booleanDictionary) Len() int { return 2 }
-
-func (d *booleanDictionary) Index(i int) Value { return makeValueBoolean(d.values[i]) }
-
-func (d *booleanDictionary) Insert(v Value) int {
-	if v.Boolean() {
-		return 1
-	} else {
-		return 0
-	}
-}
-
-func (d *booleanDictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *booleanDictionary) ReadFrom(decoder encoding.Decoder) error {
-	_, err := decoder.DecodeBoolean(d.values[:])
-	d.Reset()
-	if err != nil {
-		if err == io.EOF {
-			err = nil
-		} else {
-			err = fmt.Errorf("reading parquet dictionary of boolean values: %w", err)
-		}
-	}
-	return err
-}
-
-func (d *booleanDictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeBoolean(d.values[:]); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d boolean values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *booleanDictionary) Reset() {
-	d.values = [2]bool{false, true}
-}
-
-type int32Dictionary struct {
-	typ    Type
-	values []int32
-	index  map[int32]int32
-}
-
-func newInt32Dictionary(typ Type, bufferSize int) *int32Dictionary {
-	return &int32Dictionary{
-		typ:    typ,
-		values: make([]int32, 0, dictCap(bufferSize, 4)),
-	}
-}
-
-func (d *int32Dictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *int32Dictionary) Len() int { return len(d.values) }
-
-func (d *int32Dictionary) Index(i int) Value { return makeValueInt32(d.values[i]) }
-
-func (d *int32Dictionary) Insert(v Value) int { return d.insert(v.Int32()) }
-
-func (d *int32Dictionary) insert(value int32) int {
-	if index, exists := d.index[value]; exists {
-		return int(index)
-	}
-	if d.index == nil {
-		d.index = make(map[int32]int32, cap(d.values))
-		for i, v := range d.values {
-			d.index[v] = int32(i)
-		}
-	}
-	index := len(d.values)
-	d.index[value] = int32(index)
-	d.values = append(d.values, value)
-	return index
-}
-
-func (d *int32Dictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *int32Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]int32, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeInt32(d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+n]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of int32 values: %w", err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *int32Dictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeInt32(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d int32 values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *int32Dictionary) Reset() {
-	d.values = d.values[:0]
-	d.index = nil
-}
-
-type int64Dictionary struct {
-	typ    Type
-	values []int64
-	index  map[int64]int32
-}
-
-func newInt64Dictionary(typ Type, bufferSize int) *int64Dictionary {
-	return &int64Dictionary{
-		typ:    typ,
-		values: make([]int64, 0, dictCap(bufferSize, 8)),
-	}
-}
-
-func (d *int64Dictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *int64Dictionary) Len() int { return len(d.values) }
-
-func (d *int64Dictionary) Index(i int) Value { return makeValueInt64(d.values[i]) }
-
-func (d *int64Dictionary) Insert(v Value) int { return d.insert(v.Int64()) }
-
-func (d *int64Dictionary) insert(value int64) int {
-	if index, exists := d.index[value]; exists {
-		return int(index)
-	}
-	if d.index == nil {
-		d.index = make(map[int64]int32, cap(d.values))
-		for i, v := range d.values {
-			d.index[v] = int32(i)
-		}
-	}
-	index := len(d.values)
-	d.index[value] = int32(index)
-	d.values = append(d.values, value)
-	return index
-}
-
-func (d *int64Dictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *int64Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]int64, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeInt64(d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+n]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of int64 values: %w", err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *int64Dictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeInt64(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d int64 values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *int64Dictionary) Reset() {
-	d.values = d.values[:0]
-	d.index = nil
-}
-
-type int96Dictionary struct {
-	typ    Type
-	values []deprecated.Int96
-	index  map[deprecated.Int96]int32
-}
-
-func newInt96Dictionary(typ Type, bufferSize int) *int96Dictionary {
-	return &int96Dictionary{
-		typ:    typ,
-		values: make([]deprecated.Int96, 0, dictCap(bufferSize, 12)),
-	}
-}
-
-func (d *int96Dictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *int96Dictionary) Len() int { return len(d.values) }
-
-func (d *int96Dictionary) Index(i int) Value { return makeValueInt96(d.values[i]) }
-
-func (d *int96Dictionary) Insert(v Value) int { return d.insert(v.Int96()) }
-
-func (d *int96Dictionary) insert(value deprecated.Int96) int {
-	if index, exists := d.index[value]; exists {
-		return int(index)
-	}
-	if d.index == nil {
-		d.index = make(map[deprecated.Int96]int32, cap(d.values))
-		for i, v := range d.values {
-			d.index[v] = int32(i)
-		}
-	}
-	index := len(d.values)
-	d.index[value] = int32(index)
-	d.values = append(d.values, value)
-	return index
-}
-
-func (d *int96Dictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *int96Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]deprecated.Int96, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeInt96(d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+n]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of int96 values: %w", err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *int96Dictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeInt96(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d int96 values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *int96Dictionary) Reset() {
-	d.values = d.values[:0]
-	d.index = nil
-}
-
-type floatDictionary struct {
-	typ    Type
-	values []float32
-	index  map[float32]int32
-}
-
-func newFloatDictionary(typ Type, bufferSize int) *floatDictionary {
-	return &floatDictionary{
-		typ:    typ,
-		values: make([]float32, 0, dictCap(bufferSize, 4)),
-	}
-}
-
-func (d *floatDictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *floatDictionary) Len() int { return len(d.values) }
-
-func (d *floatDictionary) Index(i int) Value { return makeValueFloat(d.values[i]) }
-
-func (d *floatDictionary) Insert(v Value) int { return d.insert(v.Float()) }
-
-func (d *floatDictionary) insert(value float32) int {
-	if index, exists := d.index[value]; exists {
-		return int(index)
-	}
-	if d.index == nil {
-		d.index = make(map[float32]int32, cap(d.values))
-		for i, v := range d.values {
-			d.index[v] = int32(i)
-		}
-	}
-	index := len(d.values)
-	d.index[value] = int32(index)
-	d.values = append(d.values, value)
-	return index
-}
-
-func (d *floatDictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *floatDictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]float32, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeFloat(d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+n]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of float values: %w", err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *floatDictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeFloat(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d float values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *floatDictionary) Reset() {
-	d.values = d.values[:0]
-	d.index = nil
-}
-
-type doubleDictionary struct {
-	typ    Type
-	values []float64
-	index  map[float64]int32
-}
-
-func newDoubleDictionary(typ Type, bufferSize int) *doubleDictionary {
-	return &doubleDictionary{
-		typ:    typ,
-		values: make([]float64, 0, dictCap(bufferSize, 8)),
-	}
-}
-
-func (d *doubleDictionary) Type() Type { return newIndexedType(d.typ, d) }
-
-func (d *doubleDictionary) Len() int { return len(d.values) }
-
-func (d *doubleDictionary) Index(i int) Value { return makeValueDouble(d.values[i]) }
-
-func (d *doubleDictionary) Insert(v Value) int { return d.insert(v.Double()) }
-
-func (d *doubleDictionary) insert(value float64) int {
-	if index, exists := d.index[value]; exists {
-		return int(index)
-	}
-	if d.index == nil {
-		d.index = make(map[float64]int32, cap(d.values))
-		for i, v := range d.values {
-			d.index[v] = int32(i)
-		}
-	}
-	index := len(d.values)
-	d.index[value] = int32(index)
-	d.values = append(d.values, value)
-	return index
-}
-
-func (d *doubleDictionary) Lookup(indexes []int32, values []Value) {
-	for i, j := range indexes {
-		values[i] = d.Index(int(j))
-	}
-}
-
-func (d *doubleDictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]float64, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeDouble(d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+n]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of double values: %w", err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *doubleDictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeDouble(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d double values: %w", d.Len(), err)
-	}
-	return nil
-}
-
-func (d *doubleDictionary) Reset() {
-	d.values = d.values[:0]
-	d.index = nil
-}
-
 type byteArrayDictionary struct {
 	typ    Type
 	values encoding.ByteArrayList
@@ -533,28 +82,63 @@ func (d *byteArrayDictionary) Type() Type { return newIndexedType(d.typ, d) }
 
 func (d *byteArrayDictionary) Len() int { return d.values.Len() }
 
-func (d *byteArrayDictionary) Index(i int) Value { return makeValueBytes(ByteArray, d.values.Index(i)) }
+func (d *byteArrayDictionary) Index(i int32) Value {
+	return makeValueBytes(ByteArray, d.values.Index(int(i)))
+}
 
-func (d *byteArrayDictionary) Insert(v Value) int { return d.insert(v.ByteArray()) }
+func (d *byteArrayDictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
 
-func (d *byteArrayDictionary) insert(value []byte) int {
-	if index, exists := d.index[string(value)]; exists {
-		return int(index)
-	}
-	d.values.Push(value)
-	index := d.values.Len() - 1
-	stringValue := bits.BytesToString(d.values.Index(index))
 	if d.index == nil {
+		index := int32(0)
 		d.index = make(map[string]int32, d.values.Cap())
+		d.values.Range(func(v []byte) bool {
+			d.index[bits.BytesToString(v)] = index
+			index++
+			return true
+		})
+	}
+
+	for i, v := range values {
+		value := v.ByteArray()
+
+		index, exists := d.index[string(value)]
+		if !exists {
+			d.values.Push(value)
+			index = int32(d.values.Len() - 1)
+			stringValue := bits.BytesToString(d.values.Index(int(index)))
+			d.index[stringValue] = index
+		}
+
+		indexes[i] = index
 	}
-	d.index[stringValue] = int32(index)
-	return index
 }
 
 func (d *byteArrayDictionary) Lookup(indexes []int32, values []Value) {
 	for i, j := range indexes {
-		values[i] = d.Index(int(j))
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *byteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values.Index(int(indexes[0]))
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values.Index(int(i))
+			switch {
+			case bytes.Compare(value, minValue) < 0:
+				minValue = value
+			case bytes.Compare(value, maxValue) > 0:
+				maxValue = value
+			}
+		}
+
+		min = makeValueBytes(ByteArray, minValue)
+		max = makeValueBytes(ByteArray, maxValue)
 	}
+	return min, max
 }
 
 func (d *byteArrayDictionary) ReadFrom(decoder encoding.Decoder) error {
@@ -605,38 +189,67 @@ func (d *fixedLenByteArrayDictionary) Type() Type { return newIndexedType(d.typ,
 
 func (d *fixedLenByteArrayDictionary) Len() int { return len(d.values) / d.size }
 
-func (d *fixedLenByteArrayDictionary) Index(i int) Value {
+func (d *fixedLenByteArrayDictionary) Index(i int32) Value {
 	return makeValueBytes(FixedLenByteArray, d.value(i))
 }
 
-func (d *fixedLenByteArrayDictionary) value(i int) []byte {
-	return d.values[i*d.size : (i+1)*d.size]
+func (d *fixedLenByteArrayDictionary) value(i int32) []byte {
+	return d.values[int(i)*d.size : int(i+1)*d.size]
 }
 
-func (d *fixedLenByteArrayDictionary) Insert(v Value) int {
-	return d.insert(v.ByteArray())
-}
+func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
 
-func (d *fixedLenByteArrayDictionary) insert(value []byte) int {
-	if index, exists := d.index[string(value)]; exists {
-		return int(index)
-	}
 	if d.index == nil {
 		d.index = make(map[string]int32, cap(d.values)/d.size)
+		for i, j := 0, int32(0); i < len(d.values); i += d.size {
+			d.index[bits.BytesToString(d.values[i:i+d.size])] = j
+			j++
+		}
+	}
+
+	for i, v := range values {
+		value := v.ByteArray()
+
+		index, exists := d.index[string(value)]
+		if !exists {
+			index = int32(d.Len())
+			start := len(d.values)
+			d.values = append(d.values, value...)
+			d.index[bits.BytesToString(d.values[start:])] = index
+		}
+
+		indexes[i] = index
 	}
-	i := d.Len()
-	n := len(d.values)
-	d.values = append(d.values, value...)
-	d.index[bits.BytesToString(d.values[n:])] = int32(i)
-	return i
 }
 
 func (d *fixedLenByteArrayDictionary) Lookup(indexes []int32, values []Value) {
 	for i, j := range indexes {
-		values[i] = d.Index(int(j))
+		values[i] = d.Index(j)
 	}
 }
 
+func (d *fixedLenByteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.value(indexes[0])
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.value(i)
+			switch {
+			case bytes.Compare(value, minValue) < 0:
+				minValue = value
+			case bytes.Compare(value, maxValue) > 0:
+				maxValue = value
+			}
+		}
+
+		min = makeValueBytes(FixedLenByteArray, minValue)
+		max = makeValueBytes(FixedLenByteArray, maxValue)
+	}
+	return min, max
+}
+
 func (d *fixedLenByteArrayDictionary) ReadFrom(decoder encoding.Decoder) error {
 	d.Reset()
 	for {
@@ -708,21 +321,7 @@ func (page *indexedPage) NumValues() int64 { return int64(len(page.values)) }
 func (page *indexedPage) NumNulls() int64 { return 0 }
 
 func (page *indexedPage) Bounds() (min, max Value) {
-	if len(page.values) > 0 {
-		min = page.dict.Index(int(page.values[0]))
-		max = min
-		typ := page.dict.Type()
-
-		for _, i := range page.values[1:] {
-			value := page.dict.Index(int(i))
-			switch {
-			case typ.Compare(value, min) < 0:
-				min = value
-			case typ.Compare(value, max) > 0:
-				max = value
-			}
-		}
-	}
+	min, max = page.dict.Bounds(page.values)
 	min.columnIndex = page.columnIndex
 	max.columnIndex = page.columnIndex
 	return min, max
@@ -764,7 +363,7 @@ type indexedPageReader struct {
 func (r *indexedPageReader) ReadValues(values []Value) (n int, err error) {
 	var v Value
 	for n < len(values) && r.offset < len(r.page.values) {
-		v = r.page.dict.Index(int(r.page.values[r.offset]))
+		v = r.page.dict.Index(r.page.values[r.offset])
 		v.columnIndex = r.page.columnIndex
 		values[n] = v
 		r.offset++
@@ -825,8 +424,8 @@ func (col *indexedColumnBuffer) Cap() int { return cap(col.values) }
 func (col *indexedColumnBuffer) Len() int { return len(col.values) }
 
 func (col *indexedColumnBuffer) Less(i, j int) bool {
-	u := col.dict.Index(int(col.values[i]))
-	v := col.dict.Index(int(col.values[j]))
+	u := col.dict.Index(col.values[i])
+	v := col.dict.Index(col.values[j])
 	return col.typ.Compare(u, v) < 0
 }
 
@@ -835,9 +434,18 @@ func (col *indexedColumnBuffer) Swap(i, j int) {
 }
 
 func (col *indexedColumnBuffer) WriteValues(values []Value) (int, error) {
-	for _, v := range values {
-		col.values = append(col.values, int32(col.dict.Insert(v)))
+	i := len(col.values)
+	j := len(col.values) + len(values)
+
+	if j <= cap(col.values) {
+		col.values = col.values[:j]
+	} else {
+		colValues := make([]int32, j)
+		copy(colValues, col.values)
+		col.values = colValues
 	}
+
+	col.dict.Insert(col.values[i:], values)
 	return len(values), nil
 }
 
@@ -848,8 +456,8 @@ func (col *indexedColumnBuffer) WriteRow(row Row) error {
 	if len(row) > 1 {
 		return errRowHasTooManyValues(int64(len(row)))
 	}
-	col.values = append(col.values, int32(col.dict.Insert(row[0])))
-	return nil
+	_, err := col.WriteValues(row[:1])
+	return err
 }
 
 func (col *indexedColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
@@ -859,7 +467,7 @@ func (col *indexedColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
-		v := col.dict.Index(int(col.values[index]))
+		v := col.dict.Index(col.values[index])
 		v.columnIndex = col.columnIndex
 		return append(row, v), nil
 	}
@@ -943,13 +551,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/dictionary_default.go b/dictionary_default.go
new file mode 100644
index 0000000..08da320
--- /dev/null
+++ b/dictionary_default.go
@@ -0,0 +1,686 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+// The boolean dictionary always contains two values for true and false.
+type booleanDictionary struct {
+	typ    Type
+	values [2]bool
+}
+
+func newBooleanDictionary(typ Type) *booleanDictionary {
+	return &booleanDictionary{
+		typ:    typ,
+		values: [2]bool{false, true},
+	}
+}
+
+func (d *booleanDictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *booleanDictionary) Len() int { return 2 }
+
+func (d *booleanDictionary) Index(i int32) Value { return makeValueBoolean(d.values[i]) }
+
+func (d *booleanDictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	for i, v := range values {
+		if v.Boolean() {
+			indexes[i] = 1
+		} else {
+			indexes[i] = 0
+		}
+	}
+}
+
+func (d *booleanDictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *booleanDictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case compareBool(value, minValue) < 0:
+				minValue = value
+			case compareBool(value, maxValue) > 0:
+				maxValue = value
+			}
+		}
+
+		min = makeValueBoolean(minValue)
+		max = makeValueBoolean(maxValue)
+	}
+	return min, max
+}
+
+func (d *booleanDictionary) ReadFrom(decoder encoding.Decoder) error {
+	_, err := decoder.DecodeBoolean(d.values[:])
+	d.Reset()
+	if err != nil {
+		if err == io.EOF {
+			err = nil
+		} else {
+			err = fmt.Errorf("reading parquet dictionary of boolean values: %w", err)
+		}
+	}
+	return err
+}
+
+func (d *booleanDictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeBoolean(d.values[:]); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d boolean values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *booleanDictionary) Reset() {
+	d.values = [2]bool{false, true}
+}
+
+type int32Dictionary struct {
+	typ    Type
+	values []int32
+	index  map[int32]int32
+}
+
+func newInt32Dictionary(typ Type, bufferSize int) *int32Dictionary {
+	return &int32Dictionary{
+		typ:    typ,
+		values: make([]int32, 0, dictCap(bufferSize, 4)),
+	}
+}
+
+func (d *int32Dictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *int32Dictionary) Len() int { return len(d.values) }
+
+func (d *int32Dictionary) Index(i int32) Value { return makeValueInt32(d.values[i]) }
+
+func (d *int32Dictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[int32]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := v.Int32()
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *int32Dictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *int32Dictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueInt32(minValue)
+		max = makeValueInt32(maxValue)
+	}
+	return min, max
+}
+
+func (d *int32Dictionary) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]int32, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeInt32(d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of int32 values: %w", err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *int32Dictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeInt32(d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d int32 values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *int32Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+type int64Dictionary struct {
+	typ    Type
+	values []int64
+	index  map[int64]int32
+}
+
+func newInt64Dictionary(typ Type, bufferSize int) *int64Dictionary {
+	return &int64Dictionary{
+		typ:    typ,
+		values: make([]int64, 0, dictCap(bufferSize, 8)),
+	}
+}
+
+func (d *int64Dictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *int64Dictionary) Len() int { return len(d.values) }
+
+func (d *int64Dictionary) Index(i int32) Value { return makeValueInt64(d.values[i]) }
+
+func (d *int64Dictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[int64]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := v.Int64()
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *int64Dictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *int64Dictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueInt64(minValue)
+		max = makeValueInt64(maxValue)
+	}
+	return min, max
+}
+
+func (d *int64Dictionary) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]int64, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeInt64(d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of int64 values: %w", err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *int64Dictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeInt64(d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d int64 values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *int64Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+type int96Dictionary struct {
+	typ    Type
+	values []deprecated.Int96
+	index  map[deprecated.Int96]int32
+}
+
+func newInt96Dictionary(typ Type, bufferSize int) *int96Dictionary {
+	return &int96Dictionary{
+		typ:    typ,
+		values: make([]deprecated.Int96, 0, dictCap(bufferSize, 12)),
+	}
+}
+
+func (d *int96Dictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *int96Dictionary) Len() int { return len(d.values) }
+
+func (d *int96Dictionary) Index(i int32) Value { return makeValueInt96(d.values[i]) }
+
+func (d *int96Dictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[deprecated.Int96]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := v.Int96()
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *int96Dictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *int96Dictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case value.Less(minValue):
+				minValue = value
+			case maxValue.Less(value):
+				maxValue = value
+			}
+		}
+
+		min = makeValueInt96(minValue)
+		max = makeValueInt96(maxValue)
+	}
+	return min, max
+}
+
+func (d *int96Dictionary) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]deprecated.Int96, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeInt96(d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of int96 values: %w", err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *int96Dictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeInt96(d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d int96 values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *int96Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+type floatDictionary struct {
+	typ    Type
+	values []float32
+	index  map[float32]int32
+}
+
+func newFloatDictionary(typ Type, bufferSize int) *floatDictionary {
+	return &floatDictionary{
+		typ:    typ,
+		values: make([]float32, 0, dictCap(bufferSize, 4)),
+	}
+}
+
+func (d *floatDictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *floatDictionary) Len() int { return len(d.values) }
+
+func (d *floatDictionary) Index(i int32) Value { return makeValueFloat(d.values[i]) }
+
+func (d *floatDictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[float32]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := v.Float()
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *floatDictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *floatDictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueFloat(minValue)
+		max = makeValueFloat(maxValue)
+	}
+	return min, max
+}
+
+func (d *floatDictionary) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]float32, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeFloat(d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of float values: %w", err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *floatDictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeFloat(d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d float values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *floatDictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+type doubleDictionary struct {
+	typ    Type
+	values []float64
+	index  map[float64]int32
+}
+
+func newDoubleDictionary(typ Type, bufferSize int) *doubleDictionary {
+	return &doubleDictionary{
+		typ:    typ,
+		values: make([]float64, 0, dictCap(bufferSize, 8)),
+	}
+}
+
+func (d *doubleDictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *doubleDictionary) Len() int { return len(d.values) }
+
+func (d *doubleDictionary) Index(i int32) Value { return makeValueDouble(d.values[i]) }
+
+func (d *doubleDictionary) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[float64]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := v.Double()
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *doubleDictionary) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *doubleDictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueDouble(minValue)
+		max = makeValueDouble(maxValue)
+	}
+	return min, max
+}
+
+func (d *doubleDictionary) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]float64, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeDouble(d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of double values: %w", err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *doubleDictionary) WriteTo(encoder encoding.Encoder) error {
+	if err := encoder.EncodeDouble(d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d double values: %w", d.Len(), err)
+	}
+	return nil
+}
+
+func (d *doubleDictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+type uint32Dictionary struct{ *int32Dictionary }
+
+func newUint32Dictionary(typ Type, bufferSize int) uint32Dictionary {
+	return uint32Dictionary{newInt32Dictionary(typ, bufferSize)}
+}
+
+func (d uint32Dictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d uint32Dictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := uint32(d.values[indexes[0]])
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := uint32(d.values[i])
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueInt32(int32(minValue))
+		max = makeValueInt32(int32(maxValue))
+	}
+	return min, max
+}
+
+type uint64Dictionary struct{ *int64Dictionary }
+
+func newUint64Dictionary(typ Type, bufferSize int) uint64Dictionary {
+	return uint64Dictionary{newInt64Dictionary(typ, bufferSize)}
+}
+
+func (d uint64Dictionary) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d uint64Dictionary) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := uint64(d.values[indexes[0]])
+		maxValue := minValue
+
+		for _, i := range indexes[1:] {
+			value := uint64(d.values[i])
+			switch {
+			case value < minValue:
+				minValue = value
+			case value > maxValue:
+				maxValue = value
+			}
+		}
+
+		min = makeValueInt64(int64(minValue))
+		max = makeValueInt64(int64(maxValue))
+	}
+	return min, max
+}
diff --git a/dictionary_go18.go b/dictionary_go18.go
new file mode 100644
index 0000000..c663c08
--- /dev/null
+++ b/dictionary_go18.go
@@ -0,0 +1,123 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+type dictionary[T primitive] struct {
+	class  *class[T]
+	typ    Type
+	values []T
+	index  map[T]int32
+}
+
+func newDictionary[T primitive](typ Type, bufferSize int, class *class[T]) *dictionary[T] {
+	return &dictionary[T]{
+		class:  class,
+		typ:    typ,
+		values: make([]T, 0, dictCap(bufferSize, sizeof[T]())),
+	}
+}
+
+func (d *dictionary[T]) Type() Type { return newIndexedType(d.typ, d) }
+
+func (d *dictionary[T]) Len() int { return len(d.values) }
+
+func (d *dictionary[T]) Index(index int32) Value {
+	return d.class.makeValue(d.values[index])
+}
+
+func (d *dictionary[T]) Insert(indexes []int32, values []Value) {
+	_ = indexes[:len(values)]
+
+	if d.index == nil {
+		d.index = make(map[T]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
+	for i, v := range values {
+		value := d.class.value(v)
+
+		index, exists := d.index[value]
+		if !exists {
+			index = int32(len(d.values))
+			d.values = append(d.values, value)
+			d.index[value] = index
+		}
+
+		indexes[i] = index
+	}
+}
+
+func (d *dictionary[T]) Lookup(indexes []int32, values []Value) {
+	for i, j := range indexes {
+		values[i] = d.Index(j)
+	}
+}
+
+func (d *dictionary[T]) Bounds(indexes []int32) (min, max Value) {
+	if len(indexes) > 0 {
+		minValue := d.values[indexes[0]]
+		maxValue := minValue
+		less := d.class.less
+
+		for _, i := range indexes[1:] {
+			value := d.values[i]
+			switch {
+			case less(value, minValue):
+				minValue = value
+			case less(maxValue, value):
+				maxValue = value
+			}
+		}
+
+		makeValue := d.class.makeValue
+		min = makeValue(minValue)
+		max = makeValue(maxValue)
+	}
+	return min, max
+}
+
+func (d *dictionary[T]) ReadFrom(decoder encoding.Decoder) error {
+	d.Reset()
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]T, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := d.class.decode(decoder, d.values[len(d.values):cap(d.values)])
+		if n > 0 {
+			d.values = d.values[:len(d.values)+n]
+		}
+
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			} else {
+				err = fmt.Errorf("reading parquet dictionary of %s values: %w", d.class.name, err)
+			}
+			return err
+		}
+	}
+}
+
+func (d *dictionary[T]) WriteTo(encoder encoding.Encoder) error {
+	if err := d.class.encode(encoder, d.values); err != nil {
+		return fmt.Errorf("writing parquet dictionary of %d %s values: %w", d.Len(), d.class.name, err)
+	}
+	return nil
+}
+
+func (d *dictionary[T]) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
diff --git a/file.go b/file.go
index 85e1ee5..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 {
diff --git a/offset_index.go b/offset_index.go
index f3eb0d5..a9e5411 100644
--- a/offset_index.go
+++ b/offset_index.go
@@ -23,29 +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 }
+
+type fileOffsetIndex format.OffsetIndex
+
+func (i *fileOffsetIndex) NumPages() int {
+	return len(i.PageLocations)
 }
 
-func (index *offsetIndex) Offset(i int) int64 {
-	return index.PageLocations[i].Offset
+func (i *fileOffsetIndex) Offset(j int) int64 {
+	return i.PageLocations[j].Offset
 }
 
-func (index *offsetIndex) CompressedPageSize(i int) int64 {
-	return int64(index.PageLocations[i].CompressedPageSize)
+func (i *fileOffsetIndex) CompressedPageSize(j int) int64 {
+	return int64(i.PageLocations[j].CompressedPageSize)
 }
 
-func (index *offsetIndex) FirstRowIndex(i int) int64 {
-	return index.PageLocations[i].FirstRowIndex
+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 (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 }
+
+type fixedLenByteArrayOffsetIndex struct{ page *fixedLenByteArrayPage }
 
-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 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
index db84e6b..71d5166 100644
--- a/offset_index_default.go
+++ b/offset_index_default.go
@@ -2,34 +2,58 @@
 
 package parquet
 
-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 }
+type booleanOffsetIndex struct{ page *booleanPage }
 
-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 (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 }
 
-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 }
+type int32OffsetIndex struct{ page *int32Page }
 
-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 (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 }
 
-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 }
+type int64OffsetIndex struct{ page *int64Page }
 
-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 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 }
 
-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 int96OffsetIndex struct{ page *int96Page }
 
-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 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/row_group.go b/row_group.go
index 09be9c6..ab1baeb 100644
--- a/row_group.go
+++ b/row_group.go
@@ -449,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 }
 
@@ -473,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/type.go b/type.go
index b95a53d..9757aa5 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.
@@ -75,7 +87,7 @@ type Type interface {
 	// As an optimization, the method may return the same pointer across
 	// multiple calls. Applications must treat the returned value as immutable,
 	// mutating the value will result in undefined behavior.
-	PhyiscalType() *format.Type
+	PhysicalType() *format.Type
 
 	// Returns the logical type as a *format.LogicalType value. When the logical
 	// type is unknown, the method returns nil.
@@ -187,289 +199,6 @@ var convertedTypes = [...]deprecated.ConvertedType{
 	21: deprecated.Interval,
 }
 
-type primitiveType struct{}
-
-func (t primitiveType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
-
-func (t primitiveType) LogicalType() *format.LogicalType { return nil }
-
-func (t primitiveType) ConvertedType() *deprecated.ConvertedType { return nil }
-
-type booleanType struct{ primitiveType }
-
-func (t booleanType) String() string { return "BOOLEAN" }
-
-func (t booleanType) Kind() Kind { return Boolean }
-
-func (t booleanType) Length() int { return 1 }
-
-func (t booleanType) Compare(a, b Value) int {
-	return compareBool(a.Boolean(), b.Boolean())
-}
-
-func (t booleanType) PhyiscalType() *format.Type {
-	return &physicalTypes[Boolean]
-}
-
-func (t booleanType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newBooleanColumnIndexer()
-}
-
-func (t booleanType) NewDictionary(bufferSize int) Dictionary {
-	return newBooleanDictionary(t)
-}
-
-func (t booleanType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newBooleanColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t booleanType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newBooleanColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type int32Type struct{ primitiveType }
-
-func (t int32Type) String() string { return "INT32" }
-
-func (t int32Type) Kind() Kind { return Int32 }
-
-func (t int32Type) Length() int { return 32 }
-
-func (t int32Type) Compare(a, b Value) int {
-	return compareInt32(a.Int32(), b.Int32())
-}
-
-func (t int32Type) PhyiscalType() *format.Type {
-	return &physicalTypes[Int32]
-}
-
-func (t int32Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newInt32ColumnIndexer()
-}
-
-func (t int32Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt32Dictionary(t, bufferSize)
-}
-
-func (t int32Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t int32Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type int64Type struct{ primitiveType }
-
-func (t int64Type) String() string { return "INT64" }
-
-func (t int64Type) Kind() Kind { return Int64 }
-
-func (t int64Type) Length() int { return 64 }
-
-func (t int64Type) Compare(a, b Value) int {
-	return compareInt64(a.Int64(), b.Int64())
-}
-
-func (t int64Type) PhyiscalType() *format.Type {
-	return &physicalTypes[Int64]
-}
-
-func (t int64Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newInt64ColumnIndexer()
-}
-
-func (t int64Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt64Dictionary(t, bufferSize)
-}
-
-func (t int64Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t int64Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type int96Type struct{ primitiveType }
-
-func (t int96Type) String() string { return "INT96" }
-
-func (t int96Type) Kind() Kind { return Int96 }
-
-func (t int96Type) Length() int { return 96 }
-
-func (t int96Type) Compare(a, b Value) int {
-	return compareInt96(a.Int96(), b.Int96())
-}
-
-func (t int96Type) PhyiscalType() *format.Type {
-	return &physicalTypes[Int96]
-}
-
-func (t int96Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newInt96ColumnIndexer()
-}
-
-func (t int96Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt96Dictionary(t, bufferSize)
-}
-
-func (t int96Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newInt96ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t int96Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newInt96ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type floatType struct{ primitiveType }
-
-func (t floatType) String() string { return "FLOAT" }
-
-func (t floatType) Kind() Kind { return Float }
-
-func (t floatType) Length() int { return 32 }
-
-func (t floatType) Compare(a, b Value) int {
-	return compareFloat32(a.Float(), b.Float())
-}
-
-func (t floatType) PhyiscalType() *format.Type {
-	return &physicalTypes[Float]
-}
-
-func (t floatType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newFloatColumnIndexer()
-}
-
-func (t floatType) NewDictionary(bufferSize int) Dictionary {
-	return newFloatDictionary(t, bufferSize)
-}
-
-func (t floatType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newFloatColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t floatType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newFloatColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type doubleType struct{ primitiveType }
-
-func (t doubleType) String() string { return "DOUBLE" }
-
-func (t doubleType) Kind() Kind { return Double }
-
-func (t doubleType) Length() int { return 64 }
-
-func (t doubleType) Compare(a, b Value) int {
-	return compareFloat64(a.Double(), b.Double())
-}
-
-func (t doubleType) PhyiscalType() *format.Type { return &physicalTypes[Double] }
-
-func (t doubleType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newDoubleColumnIndexer()
-}
-
-func (t doubleType) NewDictionary(bufferSize int) Dictionary {
-	return newDoubleDictionary(t, bufferSize)
-}
-
-func (t doubleType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newDoubleColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t doubleType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newDoubleColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type byteArrayType struct{ primitiveType }
-
-func (t byteArrayType) String() string { return "BYTE_ARRAY" }
-
-func (t byteArrayType) Kind() Kind { return ByteArray }
-
-func (t byteArrayType) Length() int { return 0 }
-
-func (t byteArrayType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
-}
-
-func (t byteArrayType) PhyiscalType() *format.Type {
-	return &physicalTypes[ByteArray]
-}
-
-func (t byteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newByteArrayColumnIndexer(sizeLimit)
-}
-
-func (t byteArrayType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
-}
-
-func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-type fixedLenByteArrayType struct {
-	primitiveType
-	length int
-}
-
-func (t *fixedLenByteArrayType) String() string {
-	return fmt.Sprintf("FIXED_LEN_BYTE_ARRAY(%d)", t.length)
-}
-
-func (t *fixedLenByteArrayType) Kind() Kind { return FixedLenByteArray }
-
-func (t *fixedLenByteArrayType) Length() int { return t.length }
-
-func (t *fixedLenByteArrayType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
-}
-
-func (t *fixedLenByteArrayType) PhyiscalType() *format.Type {
-	return &physicalTypes[FixedLenByteArray]
-}
-
-func (t *fixedLenByteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newFixedLenByteArrayColumnIndexer(t.length, sizeLimit)
-}
-
-func (t *fixedLenByteArrayType) NewDictionary(bufferSize int) Dictionary {
-	return newFixedLenByteArrayDictionary(t, bufferSize)
-}
-
-func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newFixedLenByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-var (
-	BooleanType   Type = booleanType{}
-	Int32Type     Type = int32Type{}
-	Int64Type     Type = int64Type{}
-	Int96Type     Type = int96Type{}
-	FloatType     Type = floatType{}
-	DoubleType    Type = doubleType{}
-	ByteArrayType Type = byteArrayType{}
-)
-
-// FixedLenByteArrayType constructs a type for fixed-length values of the given
-// size (in bytes).
-func FixedLenByteArrayType(length int) Type {
-	return &fixedLenByteArrayType{length: length}
-}
-
 // Int constructs a leaf node of signed integer logical type of the given bit
 // width.
 //
@@ -553,7 +282,7 @@ func (t *intType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *intType) PhyiscalType() *format.Type {
+func (t *intType) PhysicalType() *format.Type {
 	if t.BitWidth == 64 {
 		return &physicalTypes[Int64]
 	} else {
@@ -575,54 +304,6 @@ func (t *intType) ConvertedType() *deprecated.ConvertedType {
 	return &convertedTypes[convertedType]
 }
 
-func (t *intType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	if t.IsSigned {
-		if t.BitWidth == 64 {
-			return newInt64ColumnIndexer()
-		} else {
-			return newInt32ColumnIndexer()
-		}
-	} else {
-		if t.BitWidth == 64 {
-			return newUint64ColumnIndexer()
-		} else {
-			return newUint32ColumnIndexer()
-		}
-	}
-}
-
-func (t *intType) NewDictionary(bufferSize int) Dictionary {
-	if t.BitWidth == 64 {
-		return newInt64Dictionary(t, bufferSize)
-	} else {
-		return newInt32Dictionary(t, bufferSize)
-	}
-}
-
-func (t *intType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	if t.IsSigned {
-		if t.BitWidth == 64 {
-			return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-		} else {
-			return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-		}
-	} else {
-		if t.BitWidth == 64 {
-			return newUint64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-		} else {
-			return newUint32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-		}
-	}
-}
-
-func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	if t.BitWidth == 64 {
-		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-	} else {
-		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-	}
-}
-
 // Decimal constructs a leaf node of decimal logical type with the given
 // scale, precision, and underlying type.
 //
@@ -678,7 +359,7 @@ func (t *stringType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *stringType) PhyiscalType() *format.Type {
+func (t *stringType) PhysicalType() *format.Type {
 	return &physicalTypes[ByteArray]
 }
 
@@ -731,7 +412,7 @@ func (t *uuidType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *uuidType) PhyiscalType() *format.Type {
+func (t *uuidType) PhysicalType() *format.Type {
 	return &physicalTypes[FixedLenByteArray]
 }
 
@@ -782,7 +463,7 @@ func (t *enumType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *enumType) PhyiscalType() *format.Type {
+func (t *enumType) PhysicalType() *format.Type {
 	return &physicalTypes[ByteArray]
 }
 
@@ -835,7 +516,7 @@ func (t *jsonType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *jsonType) PhyiscalType() *format.Type {
+func (t *jsonType) PhysicalType() *format.Type {
 	return &physicalTypes[ByteArray]
 }
 
@@ -884,7 +565,7 @@ func (t *bsonType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *bsonType) PhyiscalType() *format.Type {
+func (t *bsonType) PhysicalType() *format.Type {
 	return &physicalTypes[ByteArray]
 }
 
@@ -931,7 +612,7 @@ func (t *dateType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *dateType) PhyiscalType() *format.Type { return &physicalTypes[Int32] }
+func (t *dateType) PhysicalType() *format.Type { return &physicalTypes[Int32] }
 
 func (t *dateType) LogicalType() *format.LogicalType {
 	return &format.LogicalType{Date: (*format.DateType)(t)}
@@ -941,22 +622,6 @@ func (t *dateType) ConvertedType() *deprecated.ConvertedType {
 	return &convertedTypes[deprecated.Date]
 }
 
-func (t *dateType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newInt32ColumnIndexer()
-}
-
-func (t *dateType) NewDictionary(bufferSize int) Dictionary {
-	return newInt32Dictionary(t, bufferSize)
-}
-
-func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
 // TimeUnit represents units of time in the parquet type system.
 type TimeUnit interface {
 	// Returns the precision of the time unit as a time.Duration value.
@@ -1034,7 +699,7 @@ func (t *timeType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
 }
 
-func (t *timeType) PhyiscalType() *format.Type {
+func (t *timeType) PhysicalType() *format.Type {
 	if t.Unit.Millis != nil {
 		return &physicalTypes[Int32]
 	} else {
@@ -1057,38 +722,6 @@ func (t *timeType) ConvertedType() *deprecated.ConvertedType {
 	}
 }
 
-func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	if t.Unit.Millis != nil {
-		return newInt32ColumnIndexer()
-	} else {
-		return newInt64ColumnIndexer()
-	}
-}
-
-func (t *timeType) NewDictionary(bufferSize int) Dictionary {
-	if t.Unit.Millis != nil {
-		return newInt32Dictionary(t, bufferSize)
-	} else {
-		return newInt64Dictionary(t, bufferSize)
-	}
-}
-
-func (t *timeType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	if t.Unit.Millis != nil {
-		return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-	} else {
-		return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-	}
-}
-
-func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	if t.Unit.Millis != nil {
-		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-	} else {
-		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-	}
-}
-
 // Timestamp constructs of leaf node of TIMESTAMP logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp
@@ -1108,7 +741,7 @@ func (t *timestampType) Compare(a, b Value) int { return compareInt64(a.Int64(),
 
 func (t *timestampType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
 
-func (t *timestampType) PhyiscalType() *format.Type { return &physicalTypes[Int64] }
+func (t *timestampType) PhysicalType() *format.Type { return &physicalTypes[Int64] }
 
 func (t *timestampType) LogicalType() *format.LogicalType {
 	return &format.LogicalType{Timestamp: (*format.TimestampType)(t)}
@@ -1125,22 +758,6 @@ func (t *timestampType) ConvertedType() *deprecated.ConvertedType {
 	}
 }
 
-func (t *timestampType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	return newInt64ColumnIndexer()
-}
-
-func (t *timestampType) NewDictionary(bufferSize int) Dictionary {
-	return newInt64Dictionary(t, bufferSize)
-}
-
-func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
-func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
-}
-
 // List constructs a node of LIST logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#lists
@@ -1164,7 +781,7 @@ func (t *listType) Compare(Value, Value) int { panic("cannot compare values on p
 
 func (t *listType) ColumnOrder() *format.ColumnOrder { return nil }
 
-func (t *listType) PhyiscalType() *format.Type { return nil }
+func (t *listType) PhysicalType() *format.Type { return nil }
 
 func (t *listType) LogicalType() *format.LogicalType {
 	return &format.LogicalType{List: (*format.ListType)(t)}
@@ -1218,7 +835,7 @@ func (t *mapType) Compare(Value, Value) int { panic("cannot compare values on pa
 
 func (t *mapType) ColumnOrder() *format.ColumnOrder { return nil }
 
-func (t *mapType) PhyiscalType() *format.Type { return nil }
+func (t *mapType) PhysicalType() *format.Type { return nil }
 
 func (t *mapType) LogicalType() *format.LogicalType {
 	return &format.LogicalType{Map: (*format.MapType)(t)}
@@ -1256,7 +873,7 @@ func (t *nullType) Compare(Value, Value) int { panic("cannot compare values on p
 
 func (t *nullType) ColumnOrder() *format.ColumnOrder { return nil }
 
-func (t *nullType) PhyiscalType() *format.Type { return nil }
+func (t *nullType) PhysicalType() *format.Type { return nil }
 
 func (t *nullType) LogicalType() *format.LogicalType {
 	return &format.LogicalType{Unknown: (*format.NullType)(t)}
@@ -1312,7 +929,7 @@ func (groupType) Length() int { return 0 }
 
 func (groupType) ColumnOrder() *format.ColumnOrder { return nil }
 
-func (groupType) PhyiscalType() *format.Type { return nil }
+func (groupType) PhysicalType() *format.Type { return nil }
 
 func (groupType) LogicalType() *format.LogicalType { return nil }
 
diff --git a/type_default.go b/type_default.go
new file mode 100644
index 0000000..9856e7b
--- /dev/null
+++ b/type_default.go
@@ -0,0 +1,414 @@
+//go:build !go1.18
+
+package parquet
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/format"
+)
+
+var (
+	BooleanType   Type = booleanType{}
+	Int32Type     Type = int32Type{}
+	Int64Type     Type = int64Type{}
+	Int96Type     Type = int96Type{}
+	FloatType     Type = floatType{}
+	DoubleType    Type = doubleType{}
+	ByteArrayType Type = byteArrayType{}
+)
+
+type primitiveType struct{}
+
+func (t primitiveType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
+
+func (t primitiveType) LogicalType() *format.LogicalType { return nil }
+
+func (t primitiveType) ConvertedType() *deprecated.ConvertedType { return nil }
+
+type booleanType struct{ primitiveType }
+
+func (t booleanType) String() string { return "BOOLEAN" }
+
+func (t booleanType) Kind() Kind { return Boolean }
+
+func (t booleanType) Length() int { return 1 }
+
+func (t booleanType) Compare(a, b Value) int {
+	return compareBool(a.Boolean(), b.Boolean())
+}
+
+func (t booleanType) PhysicalType() *format.Type {
+	return &physicalTypes[Boolean]
+}
+
+func (t booleanType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newBooleanColumnIndexer()
+}
+
+func (t booleanType) NewDictionary(bufferSize int) Dictionary {
+	return newBooleanDictionary(t)
+}
+
+func (t booleanType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newBooleanColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t booleanType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newBooleanColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type int32Type struct{ primitiveType }
+
+func (t int32Type) String() string { return "INT32" }
+
+func (t int32Type) Kind() Kind { return Int32 }
+
+func (t int32Type) Length() int { return 32 }
+
+func (t int32Type) Compare(a, b Value) int {
+	return compareInt32(a.Int32(), b.Int32())
+}
+
+func (t int32Type) PhysicalType() *format.Type {
+	return &physicalTypes[Int32]
+}
+
+func (t int32Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newInt32ColumnIndexer()
+}
+
+func (t int32Type) NewDictionary(bufferSize int) Dictionary {
+	return newInt32Dictionary(t, bufferSize)
+}
+
+func (t int32Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t int32Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type int64Type struct{ primitiveType }
+
+func (t int64Type) String() string { return "INT64" }
+
+func (t int64Type) Kind() Kind { return Int64 }
+
+func (t int64Type) Length() int { return 64 }
+
+func (t int64Type) Compare(a, b Value) int {
+	return compareInt64(a.Int64(), b.Int64())
+}
+
+func (t int64Type) PhysicalType() *format.Type {
+	return &physicalTypes[Int64]
+}
+
+func (t int64Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newInt64ColumnIndexer()
+}
+
+func (t int64Type) NewDictionary(bufferSize int) Dictionary {
+	return newInt64Dictionary(t, bufferSize)
+}
+
+func (t int64Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t int64Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type int96Type struct{ primitiveType }
+
+func (t int96Type) String() string { return "INT96" }
+
+func (t int96Type) Kind() Kind { return Int96 }
+
+func (t int96Type) Length() int { return 96 }
+
+func (t int96Type) Compare(a, b Value) int {
+	return compareInt96(a.Int96(), b.Int96())
+}
+
+func (t int96Type) PhysicalType() *format.Type {
+	return &physicalTypes[Int96]
+}
+
+func (t int96Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newInt96ColumnIndexer()
+}
+
+func (t int96Type) NewDictionary(bufferSize int) Dictionary {
+	return newInt96Dictionary(t, bufferSize)
+}
+
+func (t int96Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newInt96ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t int96Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt96ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type floatType struct{ primitiveType }
+
+func (t floatType) String() string { return "FLOAT" }
+
+func (t floatType) Kind() Kind { return Float }
+
+func (t floatType) Length() int { return 32 }
+
+func (t floatType) Compare(a, b Value) int {
+	return compareFloat32(a.Float(), b.Float())
+}
+
+func (t floatType) PhysicalType() *format.Type {
+	return &physicalTypes[Float]
+}
+
+func (t floatType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newFloatColumnIndexer()
+}
+
+func (t floatType) NewDictionary(bufferSize int) Dictionary {
+	return newFloatDictionary(t, bufferSize)
+}
+
+func (t floatType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newFloatColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t floatType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFloatColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type doubleType struct{ primitiveType }
+
+func (t doubleType) String() string { return "DOUBLE" }
+
+func (t doubleType) Kind() Kind { return Double }
+
+func (t doubleType) Length() int { return 64 }
+
+func (t doubleType) Compare(a, b Value) int {
+	return compareFloat64(a.Double(), b.Double())
+}
+
+func (t doubleType) PhysicalType() *format.Type { return &physicalTypes[Double] }
+
+func (t doubleType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newDoubleColumnIndexer()
+}
+
+func (t doubleType) NewDictionary(bufferSize int) Dictionary {
+	return newDoubleDictionary(t, bufferSize)
+}
+
+func (t doubleType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newDoubleColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t doubleType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newDoubleColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type byteArrayType struct{ primitiveType }
+
+func (t byteArrayType) String() string { return "BYTE_ARRAY" }
+
+func (t byteArrayType) Kind() Kind { return ByteArray }
+
+func (t byteArrayType) Length() int { return 0 }
+
+func (t byteArrayType) Compare(a, b Value) int {
+	return bytes.Compare(a.ByteArray(), b.ByteArray())
+}
+
+func (t byteArrayType) PhysicalType() *format.Type {
+	return &physicalTypes[ByteArray]
+}
+
+func (t byteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newByteArrayColumnIndexer(sizeLimit)
+}
+
+func (t byteArrayType) NewDictionary(bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, bufferSize)
+}
+
+func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type fixedLenByteArrayType struct {
+	primitiveType
+	length int
+}
+
+func (t *fixedLenByteArrayType) String() string {
+	return fmt.Sprintf("FIXED_LEN_BYTE_ARRAY(%d)", t.length)
+}
+
+func (t *fixedLenByteArrayType) Kind() Kind { return FixedLenByteArray }
+
+func (t *fixedLenByteArrayType) Length() int { return t.length }
+
+func (t *fixedLenByteArrayType) Compare(a, b Value) int {
+	return bytes.Compare(a.ByteArray(), b.ByteArray())
+}
+
+func (t *fixedLenByteArrayType) PhysicalType() *format.Type {
+	return &physicalTypes[FixedLenByteArray]
+}
+
+func (t *fixedLenByteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newFixedLenByteArrayColumnIndexer(t.length, sizeLimit)
+}
+
+func (t *fixedLenByteArrayType) NewDictionary(bufferSize int) Dictionary {
+	return newFixedLenByteArrayDictionary(t, bufferSize)
+}
+
+func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newFixedLenByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+// FixedLenByteArrayType constructs a type for fixed-length values of the given
+// size (in bytes).
+func FixedLenByteArrayType(length int) Type {
+	return &fixedLenByteArrayType{length: length}
+}
+
+func (t *intType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newInt64ColumnIndexer()
+		} else {
+			return newInt32ColumnIndexer()
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newUint64ColumnIndexer()
+		} else {
+			return newUint32ColumnIndexer()
+		}
+	}
+}
+
+func (t *intType) NewDictionary(bufferSize int) Dictionary {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newInt64Dictionary(t, bufferSize)
+		} else {
+			return newInt32Dictionary(t, bufferSize)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newUint64Dictionary(t, bufferSize)
+		} else {
+			return newUint32Dictionary(t, bufferSize)
+		}
+	}
+}
+
+func (t *intType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+		} else {
+			return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newUint64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+		} else {
+			return newUint32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+		}
+	}
+}
+
+func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	if t.BitWidth == 64 {
+		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+	} else {
+		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+	}
+}
+
+func (t *dateType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newInt32ColumnIndexer()
+}
+
+func (t *dateType) NewDictionary(bufferSize int) Dictionary {
+	return newInt32Dictionary(t, bufferSize)
+}
+
+func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	if t.Unit.Millis != nil {
+		return newInt32ColumnIndexer()
+	} else {
+		return newInt64ColumnIndexer()
+	}
+}
+
+func (t *timeType) NewDictionary(bufferSize int) Dictionary {
+	if t.Unit.Millis != nil {
+		return newInt32Dictionary(t, bufferSize)
+	} else {
+		return newInt64Dictionary(t, bufferSize)
+	}
+}
+
+func (t *timeType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	if t.Unit.Millis != nil {
+		return newInt32ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+	} else {
+		return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+	}
+}
+
+func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	if t.Unit.Millis != nil {
+		return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+	} else {
+		return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+	}
+}
+
+func (t *timestampType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newInt64ColumnIndexer()
+}
+
+func (t *timestampType) NewDictionary(bufferSize int) Dictionary {
+	return newInt64Dictionary(t, bufferSize)
+}
+
+func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newInt64ColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
diff --git a/type_go18.go b/type_go18.go
new file mode 100644
index 0000000..1a229b8
--- /dev/null
+++ b/type_go18.go
@@ -0,0 +1,265 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/format"
+)
+
+var (
+	BooleanType   Type = primitiveType[bool]{class: &boolClass}
+	Int32Type     Type = primitiveType[int32]{class: &int32Class}
+	Int64Type     Type = primitiveType[int64]{class: &int64Class}
+	Int96Type     Type = primitiveType[deprecated.Int96]{class: &int96Class}
+	FloatType     Type = primitiveType[float32]{class: &float32Class}
+	DoubleType    Type = primitiveType[float64]{class: &float64Class}
+	ByteArrayType Type = byteArrayType{}
+)
+
+type primitiveType[T primitive] struct{ class *class[T] }
+
+func (t primitiveType[T]) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
+
+func (t primitiveType[T]) PhysicalType() *format.Type { return &physicalTypes[t.class.kind] }
+
+func (t primitiveType[T]) LogicalType() *format.LogicalType { return nil }
+
+func (t primitiveType[T]) ConvertedType() *deprecated.ConvertedType { return nil }
+
+func (t primitiveType[T]) String() string { return t.class.name }
+
+func (t primitiveType[T]) Kind() Kind { return t.class.kind }
+
+func (t primitiveType[T]) Length() int { return int(t.class.bits) }
+
+func (t primitiveType[T]) Compare(a, b Value) int {
+	return t.class.compare(t.class.value(a), t.class.value(b))
+}
+
+func (t primitiveType[T]) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newColumnIndexer(t.class)
+}
+
+func (t primitiveType[T]) NewDictionary(bufferSize int) Dictionary {
+	return newDictionary(t, bufferSize, t.class)
+}
+
+func (t primitiveType[T]) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, t.class)
+}
+
+func (t primitiveType[T]) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, t.class)
+}
+
+type byteArrayType struct{}
+
+func (t byteArrayType) String() string { return "BYTE_ARRAY" }
+
+func (t byteArrayType) Kind() Kind { return ByteArray }
+
+func (t byteArrayType) Length() int { return 0 }
+
+func (t byteArrayType) Compare(a, b Value) int {
+	return bytes.Compare(a.ByteArray(), b.ByteArray())
+}
+
+func (t byteArrayType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
+
+func (t byteArrayType) LogicalType() *format.LogicalType { return nil }
+
+func (t byteArrayType) ConvertedType() *deprecated.ConvertedType { return nil }
+
+func (t byteArrayType) PhysicalType() *format.Type {
+	return &physicalTypes[ByteArray]
+}
+
+func (t byteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newByteArrayColumnIndexer(sizeLimit)
+}
+
+func (t byteArrayType) NewDictionary(bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, bufferSize)
+}
+
+func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+type fixedLenByteArrayType struct{ length int }
+
+func (t *fixedLenByteArrayType) String() string {
+	return fmt.Sprintf("FIXED_LEN_BYTE_ARRAY(%d)", t.length)
+}
+
+func (t *fixedLenByteArrayType) Kind() Kind { return FixedLenByteArray }
+
+func (t *fixedLenByteArrayType) Length() int { return t.length }
+
+func (t *fixedLenByteArrayType) Compare(a, b Value) int {
+	return bytes.Compare(a.ByteArray(), b.ByteArray())
+}
+
+func (t fixedLenByteArrayType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
+
+func (t fixedLenByteArrayType) LogicalType() *format.LogicalType { return nil }
+
+func (t fixedLenByteArrayType) ConvertedType() *deprecated.ConvertedType { return nil }
+
+func (t *fixedLenByteArrayType) PhysicalType() *format.Type { return &physicalTypes[FixedLenByteArray] }
+
+func (t *fixedLenByteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newFixedLenByteArrayColumnIndexer(t.length, sizeLimit)
+}
+
+func (t *fixedLenByteArrayType) NewDictionary(bufferSize int) Dictionary {
+	return newFixedLenByteArrayDictionary(t, bufferSize)
+}
+
+func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newFixedLenByteArrayColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
+}
+
+// FixedLenByteArrayType constructs a type for fixed-length values of the given
+// size (in bytes).
+func FixedLenByteArrayType(length int) Type { return &fixedLenByteArrayType{length: length} }
+
+func (t *intType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newColumnIndexer(&int64Class)
+		} else {
+			return newColumnIndexer(&int32Class)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newColumnIndexer(&uint64Class)
+		} else {
+			return newColumnIndexer(&uint32Class)
+		}
+	}
+}
+
+func (t *intType) NewDictionary(bufferSize int) Dictionary {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newDictionary(t, bufferSize, &int64Class)
+		} else {
+			return newDictionary(t, bufferSize, &int32Class)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newDictionary(t, bufferSize, &uint64Class)
+		} else {
+			return newDictionary(t, bufferSize, &uint32Class)
+		}
+	}
+}
+
+func (t *intType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+		} else {
+			return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &uint64Class)
+		} else {
+			return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &uint32Class)
+		}
+	}
+}
+
+func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+		} else {
+			return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &uint64Class)
+		} else {
+			return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &uint32Class)
+		}
+	}
+}
+
+func (t *dateType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newColumnIndexer(&int32Class)
+}
+
+func (t *dateType) NewDictionary(bufferSize int) Dictionary {
+	return newDictionary(t, bufferSize, &int32Class)
+}
+
+func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
+}
+
+func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+}
+
+func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	if t.Unit.Millis != nil {
+		return newColumnIndexer(&int32Class)
+	} else {
+		return newColumnIndexer(&int64Class)
+	}
+}
+
+func (t *timeType) NewDictionary(bufferSize int) Dictionary {
+	if t.Unit.Millis != nil {
+		return newDictionary(t, bufferSize, &int32Class)
+	} else {
+		return newDictionary(t, bufferSize, &int64Class)
+	}
+}
+
+func (t *timeType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	if t.Unit.Millis != nil {
+		return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
+	} else {
+		return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+	}
+}
+
+func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	if t.Unit.Millis != nil {
+		return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
+	} else {
+		return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+	}
+}
+
+func (t *timestampType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
+	return newColumnIndexer(&int64Class)
+}
+
+func (t *timestampType) NewDictionary(bufferSize int) Dictionary {
+	return newDictionary(t, bufferSize, &int64Class)
+}
+
+func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
+	return newColumnBuffer(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+}
+
+func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
+	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
+}
diff --git a/writer.go b/writer.go
index 1cea895..84e9001 100644
--- a/writer.go
+++ b/writer.go
@@ -251,7 +251,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		}
 
 		w.schemaElements = append(w.schemaElements, format.SchemaElement{
-			Type:           nodeType.PhyiscalType(),
+			Type:           nodeType.PhysicalType(),
 			TypeLength:     typeLength,
 			RepetitionType: repetitionType,
 			Name:           name,
@@ -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 9a096cd..d25d6d8 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -9,6 +9,9 @@ jobs:
         go:
         - 1.17
         - 1.18
+        tags:
+        - ''
+        - purego
 
     runs-on: ubuntu-latest
 
@@ -27,4 +30,4 @@ jobs:
       run: test -z $(find . -name '*.go' | xargs gofmt -l)
 
     - name: Run Tests
-      run: go test -race ./...
+      run: go test -race -tags=${{ matrix.tags }} ./...
diff --git a/column_buffer_test.go b/column_buffer_test.go
new file mode 100644
index 0000000..dcb8868
--- /dev/null
+++ b/column_buffer_test.go
@@ -0,0 +1,42 @@
+package parquet_test
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+)
+
+type TestStruct struct {
+	A *string `parquet:"a,optional,dict"`
+}
+
+func TestOptionalDictWriteRowGroup(t *testing.T) {
+	s := parquet.SchemaOf(&TestStruct{})
+
+	str1 := "test1"
+	str2 := "test2"
+	records := []*TestStruct{
+		{A: nil},
+		{A: &str1},
+		{A: nil},
+		{A: &str2},
+		{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)
+		}
+	}
+
+	b := bytes.NewBuffer(nil)
+	w := parquet.NewWriter(b)
+	_, err := w.WriteRowGroup(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
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/dictionary_test.go b/dictionary_test.go
new file mode 100644
index 0000000..2ceb707
--- /dev/null
+++ b/dictionary_test.go
@@ -0,0 +1,49 @@
+package parquet_test
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+)
+
+func BenchmarkDictionaryPageBounds(b *testing.B) {
+	benchmarkDictionary(b, func(b *testing.B, typ parquet.Type) {
+		values := make([]parquet.Value, 1e3)
+		f := randValueFuncOf(typ)
+		r := rand.New(rand.NewSource(0))
+
+		for i := range values {
+			values[i] = f(r)
+		}
+
+		for i := 0; i < b.N; i++ {
+			page := randDictionaryPage(typ, values)
+			page.Bounds()
+		}
+	})
+}
+
+func benchmarkDictionary(b *testing.B, do func(*testing.B, parquet.Type)) {
+	types := []parquet.Type{
+		parquet.BooleanType,
+		parquet.Int32Type,
+		parquet.Int64Type,
+		parquet.Int96Type,
+		parquet.FloatType,
+		parquet.DoubleType,
+		parquet.ByteArrayType,
+	}
+
+	for _, typ := range types {
+		b.Run(typ.String(), func(b *testing.B) { do(b, typ) })
+	}
+}
+
+func randDictionaryPage(typ parquet.Type, values []parquet.Value) parquet.BufferedPage {
+	const bufferSize = 64 * 1024
+	dict := typ.NewDictionary(4 * bufferSize)
+	buf := dict.Type().NewColumnBuffer(0, bufferSize)
+	buf.WriteValues(values)
+	return buf.Page()
+}
diff --git a/parquet_test.go b/parquet_test.go
index f6f1277..feb00a1 100644
--- a/parquet_test.go
+++ b/parquet_test.go
@@ -4,9 +4,11 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"math/rand"
 	"reflect"
 
 	"github.com/segmentio/parquet-go"
+	"github.com/segmentio/parquet-go/deprecated"
 )
 
 type Contact struct {
@@ -147,3 +149,51 @@ func makeRows(any interface{}) rows {
 	}
 	return rows(slice)
 }
+
+func randValueFuncOf(t parquet.Type) func(*rand.Rand) parquet.Value {
+	switch k := t.Kind(); k {
+	case parquet.Boolean:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(r.Float64() < 0.5)
+		}
+
+	case parquet.Int32:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(r.Int31())
+		}
+
+	case parquet.Int64:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(r.Int63())
+		}
+
+	case parquet.Int96:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(deprecated.Int96{
+				0: r.Uint32(),
+				1: r.Uint32(),
+				2: r.Uint32(),
+			})
+		}
+
+	case parquet.Float:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(r.Float32())
+		}
+
+	case parquet.Double:
+		return func(r *rand.Rand) parquet.Value {
+			return parquet.ValueOf(r.Float64())
+		}
+
+	case parquet.ByteArray:
+		return func(r *rand.Rand) parquet.Value {
+			n := r.Intn(49) + 1
+			b := make([]byte, n)
+			return parquet.ValueOf(b)
+		}
+
+	default:
+		panic("NOT IMPLEMENTED")
+	}
+}
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'")
+	}
+}