Repository: segmentio/parquet-go
Pull request #115 (2022-03-24 19:09:06)

Add new decimal primitive type
Added fixed_len_byte_array decimal option
Added timestamp and date to struct tags. 
Changed it so byte arrays are not always delta encoded if plain is set as a struct tag. Some readers such as spark vectorized reader do not support this encoding.

Would appreciate some feedback on this PR. Thanks

Also want to say thanks for this project! This performance is great


Patch:
diff --git a/buffer.go b/buffer.go
index 8698881..4496a44 100644
--- a/buffer.go
+++ b/buffer.go
@@ -59,6 +59,7 @@ func (buf *Buffer) configure(schema *Schema) {
 
 	forEachLeafColumnOf(schema, func(leaf leafColumn) {
 		nullOrdering := nullsGoLast
+		columnIndex := int(leaf.columnIndex)
 		columnType := leaf.node.Type()
 		bufferSize := buf.config.ColumnBufferSize
 		dictionary := (Dictionary)(nil)
@@ -66,11 +67,11 @@ func (buf *Buffer) configure(schema *Schema) {
 
 		if isDictionaryEncoding(encoding) {
 			bufferSize /= 2
-			dictionary = columnType.NewDictionary(bufferSize)
+			dictionary = columnType.NewDictionary(columnIndex, bufferSize)
 			columnType = dictionary.Type()
 		}
 
-		column := columnType.NewColumnBuffer(int(leaf.columnIndex), bufferSize)
+		column := columnType.NewColumnBuffer(columnIndex, bufferSize)
 		switch {
 		case leaf.maxRepetitionLevel > 0:
 			column = newRepeatedColumnBuffer(column, leaf.maxRepetitionLevel, leaf.maxDefinitionLevel, nullOrdering)
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index d61f971..c4fa1d1 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -76,7 +76,8 @@ func (col *columnBuffer[T]) Write(b []byte) (int, error) {
 	if (len(b) % sizeof[T]()) != 0 {
 		return 0, fmt.Errorf("cannot write %s values from input of size %d", col.class.name, len(b))
 	}
-	return col.WriteRequired(cast.BytesToSlice[T](b))
+	n, err := col.WriteRequired(cast.BytesToSlice[T](b))
+	return sizeof[T]() * n, err
 }
 
 func (col *columnBuffer[T]) WriteRequired(values []T) (int, error) {
diff --git a/column_index.go b/column_index.go
index d23387b..c927a54 100644
--- a/column_index.go
+++ b/column_index.go
@@ -34,6 +34,57 @@ type ColumnIndex interface {
 	IsDescending() bool
 }
 
+// NewColumnIndex constructs a ColumnIndex instance from the given parquet
+// format column index. The kind argument configures the type of values
+func NewColumnIndex(kind Kind, index *format.ColumnIndex) ColumnIndex {
+	return &formatColumnIndex{
+		kind:  kind,
+		index: index,
+	}
+}
+
+type formatColumnIndex struct {
+	kind  Kind
+	index *format.ColumnIndex
+}
+
+func (f *formatColumnIndex) NumPages() int {
+	return len(f.index.MinValues)
+}
+
+func (f *formatColumnIndex) NullCount(i int) int64 {
+	if len(f.index.NullCounts) > 0 {
+		return f.index.NullCounts[0]
+	}
+	return 0
+}
+
+func (f *formatColumnIndex) NullPage(i int) bool {
+	return len(f.index.NullPages) > 0 && f.index.NullPages[i]
+}
+
+func (f *formatColumnIndex) MinValue(i int) Value {
+	if f.NullPage(i) {
+		return Value{}
+	}
+	return f.kind.Value(f.index.MinValues[i])
+}
+
+func (f *formatColumnIndex) MaxValue(i int) Value {
+	if f.NullPage(i) {
+		return Value{}
+	}
+	return f.kind.Value(f.index.MaxValues[i])
+}
+
+func (f *formatColumnIndex) IsAscending() bool {
+	return f.index.BoundaryOrder == format.Ascending
+}
+
+func (f *formatColumnIndex) IsDescending() bool {
+	return f.index.BoundaryOrder == format.Descending
+}
+
 type emptyColumnIndex struct{}
 
 func (emptyColumnIndex) NumPages() int       { return 0 }
@@ -46,17 +97,43 @@ 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) NumPages() int {
+	return len(i.chunk.columnIndex.NullPages)
+}
+
+func (i fileColumnIndex) NullCount(j int) int64 {
+	if len(i.chunk.columnIndex.NullCounts) > 0 {
+		return i.chunk.columnIndex.NullCounts[j]
+	}
+	return 0
+}
+
+func (i fileColumnIndex) NullPage(j int) bool {
+	return len(i.chunk.columnIndex.NullPages) > 0 && i.chunk.columnIndex.NullPages[j]
+}
+
+func (i fileColumnIndex) MinValue(j int) Value {
+	if i.NullPage(j) {
+		return Value{}
+	}
+	return i.makeValue(i.chunk.columnIndex.MinValues[j])
+}
+
+func (i fileColumnIndex) MaxValue(j int) Value {
+	if i.NullPage(j) {
+		return 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)
 }
diff --git a/column_mapping.go b/column_mapping.go
new file mode 100644
index 0000000..7d69cae
--- /dev/null
+++ b/column_mapping.go
@@ -0,0 +1,106 @@
+package parquet
+
+import (
+	"fmt"
+	"strings"
+)
+
+// ColumnMapping is an interface representing types which map column paths to
+// their zero-based index in a parquet schema.
+//
+// ColumnMapping is a read-only type, and therefore can be used concurrently
+// from multiple goroutines.
+type ColumnMapping struct {
+	mapping columnMapping
+	columns [][]string
+}
+
+// ColumnIndex returns the column index of the column at the given path.
+// The path is the sequence of column names identifying a leaf column, starting
+// from the root column of the schema.
+//
+// If the path was not found in the mapping, or if it did not represent a
+// leaf column of the parquet schema, the method returns a negative value.
+func (m *ColumnMapping) ColumnIndex(path ...string) (columnIndex int) {
+	return int(m.mapping.lookup(path))
+}
+
+// ColumnPaths returns the list of column paths available in the mapping.
+//
+// The method always returns the same slice value across calls to ColumnPaths,
+// applications should treat it as immutable.
+func (m *ColumnMapping) ColumnPaths() [][]string {
+	return m.columns
+}
+
+// String returns a human-readable representation of the column mapping.
+func (m *ColumnMapping) String() string {
+	s := new(strings.Builder)
+	s.WriteByte('{')
+
+	if len(m.columns) > 0 {
+		for _, path := range m.columns {
+			fmt.Fprintf(s, "\n  % 2d => %q", m.ColumnIndex(path...), columnPath(path))
+		}
+		s.WriteByte('\n')
+	}
+
+	s.WriteByte('}')
+	return s.String()
+}
+
+// ColumnMappingOf constructs the column mapping of the given schema.
+func ColumnMappingOf(schema Node) *ColumnMapping {
+	mapping := make(columnMappingGroup)
+	columns := make([][]string, 0, 16)
+
+	forEachLeafColumnOf(schema, func(leaf leafColumn) {
+		column := make([]string, len(leaf.path))
+		copy(column, leaf.path)
+		columns = append(columns, column)
+
+		group, path := mapping, leaf.path
+
+		for len(path) > 1 {
+			columnName := path[0]
+			g, ok := group[columnName].(columnMappingGroup)
+			if !ok {
+				g = make(columnMappingGroup)
+				group[columnName] = g
+			}
+			group, path = g, path[1:]
+		}
+
+		group[path[0]] = columnMappingLeaf(leaf.columnIndex)
+	})
+
+	return &ColumnMapping{
+		mapping: mapping,
+		columns: columns,
+	}
+}
+
+type columnMapping interface {
+	lookup(path columnPath) (columnIndex int16)
+}
+
+type columnMappingGroup map[string]columnMapping
+
+func (group columnMappingGroup) lookup(path columnPath) int16 {
+	if len(path) > 0 {
+		c, ok := group[path[0]]
+		if ok {
+			return c.lookup(path[1:])
+		}
+	}
+	return -1
+}
+
+type columnMappingLeaf int16
+
+func (leaf columnMappingLeaf) lookup(path columnPath) int16 {
+	if len(path) == 0 {
+		return int16(leaf)
+	}
+	return -1
+}
diff --git a/dictionary.go b/dictionary.go
index 83de7d6..922f553 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -47,17 +47,14 @@ type Dictionary interface {
 	// 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
-	// content contains only the entries read from the decoder.
-	ReadFrom(encoding.Decoder) error
-
-	// Writes the dictionary to the encoder passed as argument.
-	WriteTo(encoding.Encoder) error
-
 	// Resets the dictionary to its initial state, removing all values.
 	Reset()
+
+	// Returns a BufferedPage representing the content of the dictionary.
+	//
+	// The returned page shares the underlying memory of the buffer, it remains
+	// valid to use until the dictionary's Reset method is called.
+	Page() BufferedPage
 }
 
 func dictCap(bufferSize, valueItemSize int) int {
@@ -66,15 +63,41 @@ func dictCap(bufferSize, valueItemSize int) int {
 }
 
 type byteArrayDictionary struct {
-	typ    Type
-	values encoding.ByteArrayList
-	index  map[string]int32
+	byteArrayPage
+	typ   Type
+	index map[string]int32
 }
 
-func newByteArrayDictionary(typ Type, bufferSize int) *byteArrayDictionary {
+func newByteArrayDictionary(typ Type, columnIndex int16, bufferSize int) *byteArrayDictionary {
 	return &byteArrayDictionary{
-		typ:    typ,
-		values: encoding.MakeByteArrayList(dictCap(bufferSize, 16)),
+		typ: typ,
+		byteArrayPage: byteArrayPage{
+			values:      encoding.MakeByteArrayList(dictCap(bufferSize, 16)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readByteArrayDictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	d := &byteArrayDictionary{
+		typ: typ,
+		byteArrayPage: byteArrayPage{
+			values:      encoding.MakeByteArrayList(atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
+	for {
+		if d.values.Len() == d.values.Cap() {
+			d.values.Grow(d.values.Len())
+		}
+		_, err := decoder.DecodeByteArray(&d.values)
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			}
+			return d, err
+		}
 	}
 }
 
@@ -141,69 +164,86 @@ func (d *byteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *byteArrayDictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if d.values.Len() == d.values.Cap() {
-			d.values.Grow(d.values.Len())
-		}
-		_, err := decoder.DecodeByteArray(&d.values)
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			}
-			return err
-		}
-	}
-}
-
-func (d *byteArrayDictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeByteArray(d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d binary values: %w", d.Len(), err)
-	}
-	return nil
-}
-
 func (d *byteArrayDictionary) Reset() {
 	d.values.Reset()
 	d.index = nil
 }
 
+func (d *byteArrayDictionary) Page() BufferedPage {
+	return &d.byteArrayPage
+}
+
 type fixedLenByteArrayDictionary struct {
-	typ    Type
-	size   int
-	values []byte
-	index  map[string]int32
+	fixedLenByteArrayPage
+	typ   Type
+	index map[string]int32
 }
 
-func newFixedLenByteArrayDictionary(typ Type, bufferSize int) *fixedLenByteArrayDictionary {
+func newFixedLenByteArrayDictionary(typ Type, columnIndex int16, bufferSize int) *fixedLenByteArrayDictionary {
 	size := typ.Length()
 	return &fixedLenByteArrayDictionary{
-		typ:    typ,
-		size:   size,
-		values: make([]byte, 0, dictCap(bufferSize, size)*size),
+		typ: typ,
+		fixedLenByteArrayPage: fixedLenByteArrayPage{
+			size:        size,
+			data:        make([]byte, 0, dictCap(bufferSize, size)*size),
+			columnIndex: columnIndex,
+		},
 	}
 }
 
+func readFixedLenByteArrayDictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	size := typ.Length()
+
+	d := &fixedLenByteArrayDictionary{
+		typ: typ,
+		fixedLenByteArrayPage: fixedLenByteArrayPage{
+			size:        size,
+			data:        make([]byte, 0, atLeastOne(numValues)*size),
+			columnIndex: columnIndex,
+		},
+	}
+
+	for {
+		if len(d.data) == cap(d.data) {
+			newValues := make([]byte, len(d.data), 2*cap(d.data))
+			copy(newValues, d.data)
+			d.data = newValues
+		}
+
+		n, err := decoder.DecodeFixedLenByteArray(d.size, d.data[len(d.data):cap(d.data)])
+		if n > 0 {
+			d.data = d.data[:len(d.data)+(n*d.size)]
+		}
+
+		if err == io.EOF {
+			return d, nil
+		}
+		if err != nil {
+			return nil, fmt.Errorf("reading parquet dictionary of fixed-length binary values of size %d: %w", d.size, err)
+		}
+	}
+
+}
+
 func (d *fixedLenByteArrayDictionary) Type() Type { return newIndexedType(d.typ, d) }
 
-func (d *fixedLenByteArrayDictionary) Len() int { return len(d.values) / d.size }
+func (d *fixedLenByteArrayDictionary) Len() int { return len(d.data) / d.size }
 
 func (d *fixedLenByteArrayDictionary) Index(i int32) Value {
 	return makeValueBytes(FixedLenByteArray, d.value(i))
 }
 
 func (d *fixedLenByteArrayDictionary) value(i int32) []byte {
-	return d.values[int(i)*d.size : int(i+1)*d.size]
+	return d.data[int(i)*d.size : int(i+1)*d.size]
 }
 
 func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
 	_ = indexes[:len(values)]
 
 	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
+		d.index = make(map[string]int32, cap(d.data)/d.size)
+		for i, j := 0, int32(0); i < len(d.data); i += d.size {
+			d.index[bits.BytesToString(d.data[i:i+d.size])] = j
 			j++
 		}
 	}
@@ -214,9 +254,9 @@ func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
 		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
+			start := len(d.data)
+			d.data = append(d.data, value...)
+			d.index[bits.BytesToString(d.data[start:])] = index
 		}
 
 		indexes[i] = index
@@ -250,43 +290,15 @@ func (d *fixedLenByteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *fixedLenByteArrayDictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
-	for {
-		if len(d.values) == cap(d.values) {
-			newValues := make([]byte, len(d.values), 2*cap(d.values))
-			copy(newValues, d.values)
-			d.values = newValues
-		}
-
-		n, err := decoder.DecodeFixedLenByteArray(d.size, d.values[len(d.values):cap(d.values)])
-		if n > 0 {
-			d.values = d.values[:len(d.values)+(n*d.size)]
-		}
-
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			} else {
-				err = fmt.Errorf("reading parquet dictionary of fixed-length binary values of size %d: %w", d.size, err)
-			}
-			return err
-		}
-	}
-}
-
-func (d *fixedLenByteArrayDictionary) WriteTo(encoder encoding.Encoder) error {
-	if err := encoder.EncodeFixedLenByteArray(d.size, d.values); err != nil {
-		return fmt.Errorf("writing parquet dictionary of %d fixed-length binary values of size %d: %w", d.Len(), d.size, err)
-	}
-	return nil
-}
-
 func (d *fixedLenByteArrayDictionary) Reset() {
-	d.values = d.values[:0]
+	d.data = d.data[:0]
 	d.index = nil
 }
 
+func (d *fixedLenByteArrayDictionary) Page() BufferedPage {
+	return &d.fixedLenByteArrayPage
+}
+
 type indexedType struct {
 	Type
 	dict Dictionary
@@ -442,7 +454,7 @@ func (col *indexedColumnBuffer) WriteValues(values []Value) (int, error) {
 	if j <= cap(col.values) {
 		col.values = col.values[:j]
 	} else {
-		colValues := make([]int32, j)
+		colValues := make([]int32, j, 2*j)
 		copy(colValues, col.values)
 		col.values = colValues
 	}
diff --git a/dictionary_default.go b/dictionary_default.go
index 08da320..7b3f915 100644
--- a/dictionary_default.go
+++ b/dictionary_default.go
@@ -12,14 +12,50 @@ import (
 
 // The boolean dictionary always contains two values for true and false.
 type booleanDictionary struct {
-	typ    Type
-	values [2]bool
+	booleanPage
+	typ   Type
+	index map[bool]int32
 }
 
-func newBooleanDictionary(typ Type) *booleanDictionary {
+func newBooleanDictionary(typ Type, columnIndex int16, bufferSize int) *booleanDictionary {
 	return &booleanDictionary{
-		typ:    typ,
-		values: [2]bool{false, true},
+		typ: typ,
+		booleanPage: booleanPage{
+			values:      make([]bool, 0, atLeastOne(bufferSize)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readBooleanDictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*booleanDictionary, error) {
+	d := &booleanDictionary{
+		typ: typ,
+		booleanPage: booleanPage{
+			values:      make([]bool, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
+	for {
+		if len(d.values) == cap(d.values) {
+			newValues := make([]bool, len(d.values), 2*cap(d.values))
+			copy(newValues, d.values)
+			d.values = newValues
+		}
+
+		n, err := decoder.DecodeBoolean(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 d, err
+		}
 	}
 }
 
@@ -32,12 +68,24 @@ func (d *booleanDictionary) Index(i int32) Value { return makeValueBoolean(d.val
 func (d *booleanDictionary) Insert(indexes []int32, values []Value) {
 	_ = indexes[:len(values)]
 
+	if d.index == nil {
+		d.index = make(map[bool]int32, cap(d.values))
+		for i, v := range d.values {
+			d.index[v] = int32(i)
+		}
+	}
+
 	for i, v := range values {
-		if v.Boolean() {
-			indexes[i] = 1
-		} else {
-			indexes[i] = 0
+		value := v.Boolean()
+
+		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
 	}
 }
 
@@ -68,40 +116,60 @@ func (d *booleanDictionary) Bounds(indexes []int32) (min, max Value) {
 	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 = d.values[:0]
+	d.index = nil
 }
 
-func (d *booleanDictionary) Reset() {
-	d.values = [2]bool{false, true}
+func (d *booleanDictionary) Page() BufferedPage {
+	return &d.booleanPage
 }
 
 type int32Dictionary struct {
-	typ    Type
-	values []int32
-	index  map[int32]int32
+	int32Page
+	typ   Type
+	index map[int32]int32
 }
 
-func newInt32Dictionary(typ Type, bufferSize int) *int32Dictionary {
+func newInt32Dictionary(typ Type, columnIndex int16, bufferSize int) *int32Dictionary {
 	return &int32Dictionary{
-		typ:    typ,
-		values: make([]int32, 0, dictCap(bufferSize, 4)),
+		typ: typ,
+		int32Page: int32Page{
+			values:      make([]int32, 0, dictCap(bufferSize, 4)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readInt32Dictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*int32Dictionary, error) {
+	d := &int32Dictionary{
+		typ: typ,
+		int32Page: int32Page{
+			values:      make([]int32, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
+	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 d, err
+		}
 	}
 }
 
@@ -162,16 +230,48 @@ func (d *int32Dictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *int32Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
+func (d *int32Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+func (d *int32Dictionary) Page() BufferedPage {
+	return &d.int32Page
+}
+
+type int64Dictionary struct {
+	int64Page
+	typ   Type
+	index map[int64]int32
+}
+
+func newInt64Dictionary(typ Type, columnIndex int16, bufferSize int) *int64Dictionary {
+	return &int64Dictionary{
+		typ: typ,
+		int64Page: int64Page{
+			values:      make([]int64, 0, dictCap(bufferSize, 8)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readInt64Dictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*int64Dictionary, error) {
+	d := &int64Dictionary{
+		typ: typ,
+		int64Page: int64Page{
+			values:      make([]int64, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
 	for {
 		if len(d.values) == cap(d.values) {
-			newValues := make([]int32, len(d.values), 2*cap(d.values))
+			newValues := make([]int64, 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)])
+		n, err := decoder.DecodeInt64(d.values[len(d.values):cap(d.values)])
 		if n > 0 {
 			d.values = d.values[:len(d.values)+n]
 		}
@@ -180,38 +280,13 @@ func (d *int32Dictionary) ReadFrom(decoder encoding.Decoder) error {
 			if err == io.EOF {
 				err = nil
 			} else {
-				err = fmt.Errorf("reading parquet dictionary of int32 values: %w", err)
+				err = fmt.Errorf("reading parquet dictionary of int64 values: %w", err)
 			}
-			return err
+			return d, 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) }
@@ -269,16 +344,48 @@ func (d *int64Dictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *int64Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
+func (d *int64Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+func (d *int64Dictionary) Page() BufferedPage {
+	return &d.int64Page
+}
+
+type int96Dictionary struct {
+	int96Page
+	typ   Type
+	index map[deprecated.Int96]int32
+}
+
+func newInt96Dictionary(typ Type, columnIndex int16, bufferSize int) *int96Dictionary {
+	return &int96Dictionary{
+		typ: typ,
+		int96Page: int96Page{
+			values:      make([]deprecated.Int96, 0, dictCap(bufferSize, 12)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readInt96Dictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*int96Dictionary, error) {
+	d := &int96Dictionary{
+		typ: typ,
+		int96Page: int96Page{
+			values:      make([]deprecated.Int96, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
 	for {
 		if len(d.values) == cap(d.values) {
-			newValues := make([]int64, len(d.values), 2*cap(d.values))
+			newValues := make([]deprecated.Int96, 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)])
+		n, err := decoder.DecodeInt96(d.values[len(d.values):cap(d.values)])
 		if n > 0 {
 			d.values = d.values[:len(d.values)+n]
 		}
@@ -287,38 +394,13 @@ func (d *int64Dictionary) ReadFrom(decoder encoding.Decoder) error {
 			if err == io.EOF {
 				err = nil
 			} else {
-				err = fmt.Errorf("reading parquet dictionary of int64 values: %w", err)
+				err = fmt.Errorf("reading parquet dictionary of int96 values: %w", err)
 			}
-			return err
+			return d, 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) }
@@ -376,16 +458,48 @@ func (d *int96Dictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *int96Dictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
+func (d *int96Dictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+func (d *int96Dictionary) Page() BufferedPage {
+	return &d.int96Page
+}
+
+type floatDictionary struct {
+	floatPage
+	typ   Type
+	index map[float32]int32
+}
+
+func newFloatDictionary(typ Type, columnIndex int16, bufferSize int) *floatDictionary {
+	return &floatDictionary{
+		typ: typ,
+		floatPage: floatPage{
+			values:      make([]float32, 0, dictCap(bufferSize, 4)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readFloatDictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*floatDictionary, error) {
+	d := &floatDictionary{
+		typ: typ,
+		floatPage: floatPage{
+			values:      make([]float32, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
 	for {
 		if len(d.values) == cap(d.values) {
-			newValues := make([]deprecated.Int96, len(d.values), 2*cap(d.values))
+			newValues := make([]float32, 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)])
+		n, err := decoder.DecodeFloat(d.values[len(d.values):cap(d.values)])
 		if n > 0 {
 			d.values = d.values[:len(d.values)+n]
 		}
@@ -394,38 +508,13 @@ func (d *int96Dictionary) ReadFrom(decoder encoding.Decoder) error {
 			if err == io.EOF {
 				err = nil
 			} else {
-				err = fmt.Errorf("reading parquet dictionary of int96 values: %w", err)
+				err = fmt.Errorf("reading parquet dictionary of float values: %w", err)
 			}
-			return err
+			return d, 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) }
@@ -483,16 +572,48 @@ func (d *floatDictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
-func (d *floatDictionary) ReadFrom(decoder encoding.Decoder) error {
-	d.Reset()
+func (d *floatDictionary) Reset() {
+	d.values = d.values[:0]
+	d.index = nil
+}
+
+func (d *floatDictionary) Page() BufferedPage {
+	return &d.floatPage
+}
+
+type doubleDictionary struct {
+	doublePage
+	typ   Type
+	index map[float64]int32
+}
+
+func newDoubleDictionary(typ Type, columnIndex int16, bufferSize int) *doubleDictionary {
+	return &doubleDictionary{
+		typ: typ,
+		doublePage: doublePage{
+			values:      make([]float64, 0, dictCap(bufferSize, 8)),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readDoubleDictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (*doubleDictionary, error) {
+	d := &doubleDictionary{
+		typ: typ,
+		doublePage: doublePage{
+			values:      make([]float64, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
 	for {
 		if len(d.values) == cap(d.values) {
-			newValues := make([]float32, len(d.values), 2*cap(d.values))
+			newValues := make([]float64, 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)])
+		n, err := decoder.DecodeDouble(d.values[len(d.values):cap(d.values)])
 		if n > 0 {
 			d.values = d.values[:len(d.values)+n]
 		}
@@ -501,38 +622,13 @@ func (d *floatDictionary) ReadFrom(decoder encoding.Decoder) error {
 			if err == io.EOF {
 				err = nil
 			} else {
-				err = fmt.Errorf("reading parquet dictionary of float values: %w", err)
+				err = fmt.Errorf("reading parquet dictionary of double values: %w", err)
 			}
-			return err
+			return d, 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) }
@@ -590,47 +686,24 @@ func (d *doubleDictionary) Bounds(indexes []int32) (min, max Value) {
 	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
 }
 
+func (d *doubleDictionary) Page() BufferedPage {
+	return &d.doublePage
+}
+
 type uint32Dictionary struct{ *int32Dictionary }
 
-func newUint32Dictionary(typ Type, bufferSize int) uint32Dictionary {
-	return uint32Dictionary{newInt32Dictionary(typ, bufferSize)}
+func newUint32Dictionary(typ Type, columnIndex int16, bufferSize int) uint32Dictionary {
+	return uint32Dictionary{newInt32Dictionary(typ, columnIndex, bufferSize)}
+}
+
+func readUint32Dictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (uint32Dictionary, error) {
+	d, err := readInt32Dictionary(typ, columnIndex, numValues, decoder)
+	return uint32Dictionary{d}, err
 }
 
 func (d uint32Dictionary) Type() Type { return newIndexedType(d.typ, d) }
@@ -656,10 +729,19 @@ func (d uint32Dictionary) Bounds(indexes []int32) (min, max Value) {
 	return min, max
 }
 
+func (d uint32Dictionary) Page() BufferedPage {
+	return uint32Page{&d.int32Page}
+}
+
 type uint64Dictionary struct{ *int64Dictionary }
 
-func newUint64Dictionary(typ Type, bufferSize int) uint64Dictionary {
-	return uint64Dictionary{newInt64Dictionary(typ, bufferSize)}
+func newUint64Dictionary(typ Type, columnIndex int16, bufferSize int) uint64Dictionary {
+	return uint64Dictionary{newInt64Dictionary(typ, columnIndex, bufferSize)}
+}
+
+func readUint64Dictionary(typ Type, columnIndex int16, numValues int, decoder encoding.Decoder) (uint64Dictionary, error) {
+	d, err := readInt64Dictionary(typ, columnIndex, numValues, decoder)
+	return uint64Dictionary{d}, err
 }
 
 func (d uint64Dictionary) Type() Type { return newIndexedType(d.typ, d) }
@@ -684,3 +766,7 @@ func (d uint64Dictionary) Bounds(indexes []int32) (min, max Value) {
 	}
 	return min, max
 }
+
+func (d uint64Dictionary) Page() BufferedPage {
+	return uint64Page{&d.int64Page}
+}
diff --git a/dictionary_go18.go b/dictionary_go18.go
index c663c08..6f68290 100644
--- a/dictionary_go18.go
+++ b/dictionary_go18.go
@@ -10,17 +10,52 @@ import (
 )
 
 type dictionary[T primitive] struct {
-	class  *class[T]
-	typ    Type
-	values []T
-	index  map[T]int32
+	page[T]
+	typ   Type
+	index map[T]int32
 }
 
-func newDictionary[T primitive](typ Type, bufferSize int, class *class[T]) *dictionary[T] {
+func newDictionary[T primitive](typ Type, columnIndex int16, bufferSize int, class *class[T]) *dictionary[T] {
 	return &dictionary[T]{
-		class:  class,
-		typ:    typ,
-		values: make([]T, 0, dictCap(bufferSize, sizeof[T]())),
+		typ: typ,
+		page: page[T]{
+			class:       class,
+			values:      make([]T, 0, dictCap(bufferSize, sizeof[T]())),
+			columnIndex: columnIndex,
+		},
+	}
+}
+
+func readDictionary[T primitive](typ Type, columnIndex int16, numValues int, decoder encoding.Decoder, class *class[T]) (*dictionary[T], error) {
+	d := &dictionary[T]{
+		typ: typ,
+		page: page[T]{
+			class:       class,
+			values:      make([]T, 0, atLeastOne(numValues)),
+			columnIndex: columnIndex,
+		},
+	}
+
+	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 d, err
+		}
 	}
 }
 
@@ -85,39 +120,11 @@ func (d *dictionary[T]) Bounds(indexes []int32) (min, max Value) {
 	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
 }
+
+func (d *dictionary[T]) Page() BufferedPage {
+	return &d.page
+}
diff --git a/file.go b/file.go
index 7035911..4a99839 100644
--- a/file.go
+++ b/file.go
@@ -17,6 +17,7 @@ import (
 )
 
 const (
+	defaultDictBufferSize  = 8192
 	defaultReadBufferSize  = 4096
 	defaultLevelBufferSize = 1024
 )
@@ -205,6 +206,9 @@ func (f *File) readPageIndex(section *bufferedSectionReader, decoder *thrift.Dec
 	return columnIndexes, offsetIndexes, nil
 }
 
+// NumRows returns the number of rows in the file.
+func (f *File) NumRows() int64 { return f.metadata.NumRows }
+
 // NumRowGroups returns the number of row groups in f.
 func (f *File) NumRowGroups() int { return len(f.rowGroups) }
 
@@ -541,8 +545,6 @@ func (r *filePages) readPage() (*filePage, error) {
 }
 
 func (r *filePages) readDictionary() error {
-	r.page.dictionary = r.page.columnType.NewDictionary(0) // TODO: configure buffer size
-	r.page.columnType = r.page.dictionary.Type()
 	currentOffset, _ := r.section.Seek(0, io.SeekCurrent)
 	defer func() {
 		r.section.Seek(currentOffset, io.SeekStart)
@@ -559,19 +561,20 @@ func (r *filePages) readDictionary() error {
 		return err
 	}
 
-	dict := r.page.dictionary
 	page := acquireCompressedPageReader(p.codec, &p.data)
-
 	enc := r.page.header.DictionaryPageHeader.Encoding
 	dec := LookupEncoding(enc).NewDecoder(page)
 
-	dict.Reset()
-	err = dict.ReadFrom(dec)
+	columnIndex := r.column.Column()
+	numValues := int(p.NumValues())
+	dict, err := r.page.columnType.ReadDictionary(columnIndex, numValues, dec)
 	releaseCompressedPageReader(page)
 
 	if err != nil {
 		return fmt.Errorf("reading dictionary of column %q: %w", p.columnPath(), err)
 	}
+	r.page.dictionary = dict
+	r.page.columnType = dict.Type()
 	return nil
 }
 
diff --git a/node.go b/node.go
index 27a2ce2..315ba8f 100644
--- a/node.go
+++ b/node.go
@@ -496,6 +496,7 @@ func encodingAndCompressionOf(node Node) (encoding.Encoding, compress.Codec) {
 	// representations of the pages, picking the one with the smallest space
 	// footprint; keep it simple for now.
 	encoding := encoding.Encoding(&Plain)
+	nodeEncoding := node.Encoding()
 	compression := compress.Codec(&Uncompressed)
 	// The parquet-format documentation states that the
 	// DELTA_LENGTH_BYTE_ARRAY is always preferred to PLAIN when
@@ -504,11 +505,11 @@ func encodingAndCompressionOf(node Node) (encoding.Encoding, compress.Codec) {
 	// the opportunity to override this behavior if needed.
 	//
 	// https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-length-byte-array-delta_length_byte_array--6
-	if node.Type().Kind() == ByteArray {
+	if node.Type().Kind() == ByteArray && len(nodeEncoding) == 0 {
 		encoding = &DeltaLengthByteArray
 	}
 
-	for _, e := range node.Encoding() {
+	for _, e := range nodeEncoding {
 		encoding = e
 		break
 	}
diff --git a/page_go18.go b/page_go18.go
index b398994..4c9dbd5 100644
--- a/page_go18.go
+++ b/page_go18.go
@@ -74,7 +74,8 @@ type pageReader[T primitive] struct {
 }
 
 func (r *pageReader[T]) Read(b []byte) (n int, err error) {
-	return r.ReadRequired(cast.BytesToSlice[T](b))
+	n, err = r.ReadRequired(cast.BytesToSlice[T](b))
+	return sizeof[T]() * n, err
 }
 
 func (r *pageReader[T]) ReadRequired(values []T) (n int, err error) {
diff --git a/reader.go b/reader.go
index b354cef..79274c7 100644
--- a/reader.go
+++ b/reader.go
@@ -95,19 +95,50 @@ func NewReader(input io.ReaderAt, options ...ReaderOption) *Reader {
 		r.file.rowGroup = concat(schema, rowGroups)
 	}
 
-	if c.Schema != nil && !nodesAreEqual(c.Schema, r.file.schema) {
-		conv, err := Convert(c.Schema, r.file.schema)
+	if c.Schema != nil {
+		r.file.schema = c.Schema
+		r.file.rowGroup = convertRowGroupTo(r.file.rowGroup, c.Schema)
+	}
+
+	r.read.init(r.file.schema, r.file.rowGroup)
+	return r
+}
+
+// NewRowGroupReader constructs a new Reader which reads rows from the RowGroup
+// passed as argument.
+func NewRowGroupReader(rowGroup RowGroup, options ...ReaderOption) *Reader {
+	c, err := NewReaderConfig(options...)
+	if err != nil {
+		panic(err)
+	}
+
+	if c.Schema != nil {
+		rowGroup = convertRowGroupTo(rowGroup, c.Schema)
+	}
+
+	r := &Reader{
+		file: reader{
+			schema:   rowGroup.Schema(),
+			rowGroup: rowGroup,
+		},
+	}
+
+	r.read.init(r.file.schema, r.file.rowGroup)
+	return r
+}
+
+func convertRowGroupTo(rowGroup RowGroup, schema *Schema) RowGroup {
+	if rowGroupSchema := rowGroup.Schema(); !nodesAreEqual(schema, rowGroupSchema) {
+		conv, err := Convert(schema, rowGroupSchema)
 		if err != nil {
 			// TODO: this looks like something we should not be panicking on,
 			// but the current NewReader API does not offer a mechanism to
 			// report errors.
 			panic(err)
 		}
-		r.file.rowGroup = ConvertRowGroup(r.file.rowGroup, conv)
+		rowGroup = ConvertRowGroup(rowGroup, conv)
 	}
-
-	r.read.init(r.file.schema, r.file.rowGroup)
-	return r
+	return rowGroup
 }
 
 func sizeOf(r io.ReaderAt) (int64, error) {
diff --git a/schema.go b/schema.go
index b0419fe..cc3d4d7 100644
--- a/schema.go
+++ b/schema.go
@@ -40,19 +40,23 @@ type Schema struct {
 //
 // The following options are also supported in the "parquet" struct tag:
 //
-//	optional | make the parquet column optional
-//	snappy   | sets the parquet column compression codec to snappy
-//	gzip     | sets the parquet column compression codec to gzip
-//	brotli   | sets the parquet column compression codec to brotli
-//	lz4      | sets the parquet column compression codec to lz4
-//	zstd     | sets the parquet column compression codec to zstd
-//	plain    | enables the plain encoding (no-op default)
-//	dict     | enables dictionary encoding on the parquet column
-//	delta    | enables delta encoding on the parquet column
-//	list     | for slice types, use the parquet LIST logical type
-//	enum     | for string types, use the parquet ENUM logical type
-//	uuid     | for string and [16]byte types, use the parquet UUID logical type
-//	decimal  | for int32 and int64 types, use the parquet DECIMAL logical type
+//	optional  | make the parquet column optional
+//	snappy    | sets the parquet column compression codec to snappy
+//	gzip      | sets the parquet column compression codec to gzip
+//	brotli    | sets the parquet column compression codec to brotli
+//	lz4       | sets the parquet column compression codec to lz4
+//	zstd      | sets the parquet column compression codec to zstd
+//	plain     | enables the plain encoding (no-op default)
+//	dict      | enables dictionary encoding on the parquet column
+//	delta     | enables delta encoding on the parquet column
+//	list      | for slice types, use the parquet LIST logical type
+//	enum      | for string types, use the parquet ENUM logical type
+//	uuid      | for string and [16]byte types, use the parquet UUID logical type
+//	decimal   | for int32, int64 and [n]byte types, use the parquet DECIMAL logical type
+//	date      | for int32 types use the DATE logical type
+//	timestamp | for int64 types use the TIMESTAMP logical type with millisecond precision
+//
+// The date logical type is an int32 value of the number of days since the unix epoch
 //
 // The decimal tag must be followed by two integer parameters, the first integer
 // representing the scale and the second the precision; for example:
@@ -552,11 +556,27 @@ func makeStructField(f reflect.StructField) structField {
 					baseType = Int32Type
 				case reflect.Int64:
 					baseType = Int64Type
+				case reflect.Array:
+					baseType = FixedLenByteArrayType(calcDecimalFixedLenByteArraySize(precision))
 				default:
 					throwInvalidFieldTag(f, option)
 				}
-				setNode(Decimal(scale, precision, baseType))
 
+				setNode(Decimal(scale, precision, baseType))
+			case "date":
+				switch f.Type.Kind() {
+				case reflect.Int32:
+					setNode(Date())
+				default:
+					throwInvalidFieldTag(f, option)
+				}
+			case "timestamp":
+				switch f.Type.Kind() {
+				case reflect.Int64:
+					setNode(Timestamp(Millisecond))
+				default:
+					throwInvalidFieldTag(f, option)
+				}
 			default:
 				throwUnknownFieldTag(f, option)
 			}
diff --git a/search.go b/search.go
new file mode 100644
index 0000000..758b1b3
--- /dev/null
+++ b/search.go
@@ -0,0 +1,113 @@
+package parquet
+
+// CompareNullsFirst constructs a comparison function which assumes that null
+// values are smaller than all other values.
+func CompareNullsFirst(cmp func(Value, Value) int) func(Value, Value) int {
+	return func(a, b Value) int {
+		switch {
+		case a.IsNull():
+			if b.IsNull() {
+				return 0
+			}
+			return -1
+		case b.IsNull():
+			return +1
+		default:
+			return cmp(a, b)
+		}
+	}
+}
+
+// CompareNullsLast constructs a comparison function which assumes that null
+// values are greater than all other values.
+func CompareNullsLast(cmp func(Value, Value) int) func(Value, Value) int {
+	return func(a, b Value) int {
+		switch {
+		case a.IsNull():
+			if b.IsNull() {
+				return 0
+			}
+			return +1
+		case b.IsNull():
+			return -1
+		default:
+			return cmp(a, b)
+		}
+	}
+}
+
+// Search is like Find, but uses the default ordering of the given type.
+func Search(index ColumnIndex, value Value, typ Type) int {
+	return Find(index, value, CompareNullsLast(typ.Compare))
+}
+
+// Find uses the column index passed as argument to find the page that the
+// given value is expected to be found in.
+//
+// The function returns the index of the first page that might contain the
+// value. If the function determines that the value does not exist in the
+// index, NumPages is returned.
+//
+// The comparison function passed as last argument is used to determine the
+// relative order of values. This should generally be the Compare method of
+// the column type, but can sometimes be customized to modify how null values
+// are interpreted, for example:
+//
+//	pageIndex := parquet.Find(columnIndex, value,
+//		parquet.CompareNullsFirst(typ.Compare),
+//	)
+//
+func Find(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
+	switch {
+	case index.IsAscending():
+		return binarySearch(index, value, cmp)
+	default:
+		return linearSearch(index, value, cmp)
+	}
+}
+
+func binarySearch(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
+	n := index.NumPages()
+	i := 0
+	j := n
+
+	for (j - i) > 1 {
+		k := ((j - i) / 2) + i
+		c := cmp(value, index.MinValue(k))
+
+		switch {
+		case c < 0:
+			j = k
+		case c > 0:
+			i = k
+		default:
+			return k
+		}
+	}
+
+	if i < n {
+		min := index.MinValue(i)
+		max := index.MaxValue(i)
+
+		if cmp(value, min) < 0 || cmp(max, value) < 0 {
+			i = n
+		}
+	}
+
+	return i
+}
+
+func linearSearch(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
+	n := index.NumPages()
+
+	for i := 0; i < n; i++ {
+		min := index.MinValue(i)
+		max := index.MaxValue(i)
+
+		if cmp(min, value) <= 0 && cmp(value, max) <= 0 {
+			return i
+		}
+	}
+
+	return n
+}
diff --git a/type.go b/type.go
index 9757aa5..4b2e6ce 100644
--- a/type.go
+++ b/type.go
@@ -3,11 +3,13 @@ package parquet
 import (
 	"bytes"
 	"fmt"
+	"math"
 	"reflect"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
@@ -30,7 +32,7 @@ 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.
+// Value constructs a value from 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.
@@ -119,7 +121,7 @@ type Type interface {
 	// Creates a dictionary holding values of this type.
 	//
 	// The method panics if it is called on a group type.
-	NewDictionary(bufferSize int) Dictionary
+	NewDictionary(columnIndex, bufferSize int) Dictionary
 
 	// Creates a row group buffer column for values of this type.
 	//
@@ -155,6 +157,14 @@ type Type interface {
 	// parquet.Int32Reader interface to allow programs to more efficiently read
 	// columns of INT32 values.
 	NewColumnReader(columnIndex, bufferSize int) ColumnReader
+
+	// Reads a dictionary with values of this type from the decoder passed as
+	// argument.
+	//
+	// The number of values is a hint to optimize the allocation of memory
+	// buffers for the dictionary. Callers that don't know how many values will
+	// be decoded should pass zero for numValues.
+	ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error)
 }
 
 // In the current parquet version supported by this library, only type-defined
@@ -304,15 +314,21 @@ func (t *intType) ConvertedType() *deprecated.ConvertedType {
 	return &convertedTypes[convertedType]
 }
 
+// FixedLenByteArray decimals are sized based on precision
+// this function calculates the necessary byte array size
+func calcDecimalFixedLenByteArraySize(precision int) int {
+	return int(math.Ceil((math.Log10(2) + float64(precision)) / math.Log10(256)))
+}
+
 // Decimal constructs a leaf node of decimal logical type with the given
 // scale, precision, and underlying type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#decimal
 func Decimal(scale, precision int, typ Type) Node {
-	switch typ {
-	case Int32Type, Int64Type:
+	switch typ.Kind() {
+	case Int32, Int64, FixedLenByteArray:
 	default:
-		panic("DECIMAL node must annotate the INT32 or INT64 types but got " + typ.String())
+		panic("DECIMAL node must annotate Int32, Int64 or FixedLenByteArray but got " + typ.String())
 	}
 	return Leaf(&decimalType{
 		decimal: format.DecimalType{
@@ -375,8 +391,8 @@ func (t *stringType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t *stringType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t *stringType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *stringType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -387,6 +403,10 @@ func (t *stringType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *stringType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 func (t *stringType) GoType() reflect.Type {
 	return reflect.TypeOf("")
 }
@@ -426,8 +446,8 @@ func (t *uuidType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newFixedLenByteArrayColumnIndexer(16, sizeLimit)
 }
 
-func (t *uuidType) NewDictionary(bufferSize int) Dictionary {
-	return newFixedLenByteArrayDictionary(t, bufferSize)
+func (t *uuidType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *uuidType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -438,6 +458,10 @@ func (t *uuidType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *uuidType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 func (t *uuidType) GoType() reflect.Type {
 	return reflect.TypeOf(uuid.UUID{})
 }
@@ -479,8 +503,8 @@ func (t *enumType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t *enumType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t *enumType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *enumType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -491,6 +515,10 @@ func (t *enumType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *enumType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 func (t *enumType) GoType() reflect.Type {
 	return reflect.TypeOf("")
 }
@@ -532,8 +560,8 @@ func (t *jsonType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t *jsonType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t *jsonType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *jsonType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -544,6 +572,10 @@ func (t *jsonType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *jsonType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 // BSON constructs a leaf node of BSON logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#bson
@@ -581,8 +613,8 @@ func (t *bsonType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t *bsonType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t *bsonType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *bsonType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -593,6 +625,10 @@ func (t *bsonType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *bsonType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 // Date constructs a leaf node of DATE logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#date
@@ -667,12 +703,20 @@ func Time(unit TimeUnit) Node {
 
 type timeType format.TimeType
 
+func (t *timeType) useInt32() bool {
+	return t.Unit.Millis != nil
+}
+
+func (t *timeType) useInt64() bool {
+	return t.Unit.Micros != nil
+}
+
 func (t *timeType) String() string {
 	return (*format.TimeType)(t).String()
 }
 
 func (t *timeType) Kind() Kind {
-	if t.Unit.Millis != nil {
+	if t.useInt32() {
 		return Int32
 	} else {
 		return Int64
@@ -680,7 +724,7 @@ func (t *timeType) Kind() Kind {
 }
 
 func (t *timeType) Length() int {
-	if t.Unit.Millis != nil {
+	if t.useInt32() {
 		return 32
 	} else {
 		return 64
@@ -688,7 +732,7 @@ func (t *timeType) Length() int {
 }
 
 func (t *timeType) Compare(a, b Value) int {
-	if t.Unit.Millis != nil {
+	if t.useInt32() {
 		return compareInt32(a.Int32(), b.Int32())
 	} else {
 		return compareInt64(a.Int64(), b.Int64())
@@ -700,7 +744,7 @@ func (t *timeType) ColumnOrder() *format.ColumnOrder {
 }
 
 func (t *timeType) PhysicalType() *format.Type {
-	if t.Unit.Millis != nil {
+	if t.useInt32() {
 		return &physicalTypes[Int32]
 	} else {
 		return &physicalTypes[Int64]
@@ -713,9 +757,9 @@ func (t *timeType) LogicalType() *format.LogicalType {
 
 func (t *timeType) ConvertedType() *deprecated.ConvertedType {
 	switch {
-	case t.Unit.Millis != nil:
+	case t.useInt32():
 		return &convertedTypes[deprecated.TimeMillis]
-	case t.Unit.Micros != nil:
+	case t.useInt64():
 		return &convertedTypes[deprecated.TimeMicros]
 	default:
 		return nil
@@ -791,20 +835,24 @@ func (t *listType) ConvertedType() *deprecated.ConvertedType {
 	return &convertedTypes[deprecated.List]
 }
 
-func (t *listType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	panic("create create a column indexer from parquet LIST type")
+func (t *listType) NewColumnIndexer(int) ColumnIndexer {
+	panic("create create column indexer from parquet LIST type")
 }
 
-func (t *listType) NewDictionary(bufferSize int) Dictionary {
+func (t *listType) NewDictionary(int, int) Dictionary {
 	panic("cannot create dictionary from parquet LIST type")
 }
 
-func (t *listType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	panic("cannot create row group column from parquet LIST type")
+func (t *listType) NewColumnBuffer(int, int) ColumnBuffer {
+	panic("cannot create column buffer from parquet LIST type")
+}
+
+func (t *listType) NewColumnReader(int, int) ColumnReader {
+	panic("cannot create column reader from parquet LIST type")
 }
 
-func (t *listType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	panic("cannot create row group column from parquet LIST type")
+func (t *listType) ReadDictionary(int, int, encoding.Decoder) (Dictionary, error) {
+	panic("cannot read dictionary from parquet LIST type")
 }
 
 // Map constructs a node of MAP logical type.
@@ -845,20 +893,24 @@ func (t *mapType) ConvertedType() *deprecated.ConvertedType {
 	return &convertedTypes[deprecated.Map]
 }
 
-func (t *mapType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
-	panic("create create a column indexer from parquet MAP type")
+func (t *mapType) NewColumnIndexer(int) ColumnIndexer {
+	panic("create create column indexer from parquet MAP type")
 }
 
-func (t *mapType) NewDictionary(bufferSize int) Dictionary {
+func (t *mapType) NewDictionary(int, int) Dictionary {
 	panic("cannot create dictionary from parquet MAP type")
 }
 
-func (t *mapType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
-	panic("cannot create row group column from parquet MAP type")
+func (t *mapType) NewColumnBuffer(int, int) ColumnBuffer {
+	panic("cannot create column buffer from parquet MAP type")
+}
+
+func (t *mapType) NewColumnReader(int, int) ColumnReader {
+	panic("cannot create column reader from parquet MAP type")
 }
 
-func (t *mapType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
-	panic("cannot create row group column from parquet MAP type")
+func (t *mapType) ReadDictionary(int, int, encoding.Decoder) (Dictionary, error) {
+	panic("cannot read dictionary from parquet MAP type")
 }
 
 type nullType format.NullType
@@ -882,19 +934,23 @@ func (t *nullType) LogicalType() *format.LogicalType {
 func (t *nullType) ConvertedType() *deprecated.ConvertedType { return nil }
 
 func (t *nullType) NewColumnIndexer(int) ColumnIndexer {
-	panic("create create a column indexer from parquet NULL type")
+	panic("create create column indexer from parquet NULL type")
 }
 
-func (t *nullType) NewDictionary(int) Dictionary {
+func (t *nullType) NewDictionary(int, int) Dictionary {
 	panic("cannot create dictionary from parquet NULL type")
 }
 
 func (t *nullType) NewColumnBuffer(int, int) ColumnBuffer {
-	panic("cannot create row group column from parquet NULL type")
+	panic("cannot create column buffer from parquet NULL type")
 }
 
 func (t *nullType) NewColumnReader(int, int) ColumnReader {
-	panic("cannot create row group column from parquet NULL type")
+	panic("cannot create column reader from parquet NULL type")
+}
+
+func (t *nullType) ReadDictionary(int, int, encoding.Decoder) (Dictionary, error) {
+	panic("cannot read dictionary from parquet NULL type")
 }
 
 type groupType struct{}
@@ -913,16 +969,20 @@ func (groupType) NewColumnIndexer(int) ColumnIndexer {
 	panic("cannot create column indexer from parquet group")
 }
 
-func (groupType) NewDictionary(int) Dictionary {
+func (groupType) NewDictionary(int, int) Dictionary {
 	panic("cannot create dictionary from parquet group")
 }
 
 func (t groupType) NewColumnBuffer(int, int) ColumnBuffer {
-	panic("cannot create row group column from parquet group")
+	panic("cannot create column buffer from parquet group")
 }
 
 func (t groupType) NewColumnReader(int, int) ColumnReader {
-	panic("cannot create row group column from parquet group")
+	panic("cannot create column reader from parquet group")
+}
+
+func (t groupType) ReadDictionary(int, int, encoding.Decoder) (Dictionary, error) {
+	panic("cannot read dictionary from parquet group")
 }
 
 func (groupType) Length() int { return 0 }
diff --git a/type_default.go b/type_default.go
index 9856e7b..0961cd1 100644
--- a/type_default.go
+++ b/type_default.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -48,8 +49,8 @@ func (t booleanType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newBooleanColumnIndexer()
 }
 
-func (t booleanType) NewDictionary(bufferSize int) Dictionary {
-	return newBooleanDictionary(t)
+func (t booleanType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newBooleanDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t booleanType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -60,6 +61,10 @@ func (t booleanType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newBooleanColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t booleanType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readBooleanDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type int32Type struct{ primitiveType }
 
 func (t int32Type) String() string { return "INT32" }
@@ -80,8 +85,8 @@ func (t int32Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newInt32ColumnIndexer()
 }
 
-func (t int32Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt32Dictionary(t, bufferSize)
+func (t int32Type) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newInt32Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t int32Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -92,6 +97,10 @@ func (t int32Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t int32Type) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readInt32Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type int64Type struct{ primitiveType }
 
 func (t int64Type) String() string { return "INT64" }
@@ -112,8 +121,8 @@ func (t int64Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newInt64ColumnIndexer()
 }
 
-func (t int64Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt64Dictionary(t, bufferSize)
+func (t int64Type) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newInt64Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t int64Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -124,6 +133,10 @@ func (t int64Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t int64Type) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readInt64Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type int96Type struct{ primitiveType }
 
 func (t int96Type) String() string { return "INT96" }
@@ -144,8 +157,8 @@ func (t int96Type) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newInt96ColumnIndexer()
 }
 
-func (t int96Type) NewDictionary(bufferSize int) Dictionary {
-	return newInt96Dictionary(t, bufferSize)
+func (t int96Type) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newInt96Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t int96Type) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -156,6 +169,10 @@ func (t int96Type) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newInt96ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t int96Type) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readInt96Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type floatType struct{ primitiveType }
 
 func (t floatType) String() string { return "FLOAT" }
@@ -176,8 +193,8 @@ func (t floatType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newFloatColumnIndexer()
 }
 
-func (t floatType) NewDictionary(bufferSize int) Dictionary {
-	return newFloatDictionary(t, bufferSize)
+func (t floatType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newFloatDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t floatType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -188,6 +205,10 @@ func (t floatType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newFloatColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t floatType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readFloatDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type doubleType struct{ primitiveType }
 
 func (t doubleType) String() string { return "DOUBLE" }
@@ -206,8 +227,8 @@ func (t doubleType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newDoubleColumnIndexer()
 }
 
-func (t doubleType) NewDictionary(bufferSize int) Dictionary {
-	return newDoubleDictionary(t, bufferSize)
+func (t doubleType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newDoubleDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t doubleType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -218,6 +239,10 @@ func (t doubleType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newDoubleColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t doubleType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readDoubleDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type byteArrayType struct{ primitiveType }
 
 func (t byteArrayType) String() string { return "BYTE_ARRAY" }
@@ -238,8 +263,8 @@ func (t byteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t byteArrayType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t byteArrayType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -250,6 +275,10 @@ func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t byteArrayType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type fixedLenByteArrayType struct {
 	primitiveType
 	length int
@@ -275,8 +304,8 @@ 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) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -287,6 +316,10 @@ func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) Col
 	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *fixedLenByteArrayType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 // FixedLenByteArrayType constructs a type for fixed-length values of the given
 // size (in bytes).
 func FixedLenByteArrayType(length int) Type {
@@ -309,18 +342,18 @@ func (t *intType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	}
 }
 
-func (t *intType) NewDictionary(bufferSize int) Dictionary {
+func (t *intType) NewDictionary(columnIndex, bufferSize int) Dictionary {
 	if t.IsSigned {
 		if t.BitWidth == 64 {
-			return newInt64Dictionary(t, bufferSize)
+			return newInt64Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 		} else {
-			return newInt32Dictionary(t, bufferSize)
+			return newInt32Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 		}
 	} else {
 		if t.BitWidth == 64 {
-			return newUint64Dictionary(t, bufferSize)
+			return newUint64Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 		} else {
-			return newUint32Dictionary(t, bufferSize)
+			return newUint32Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 		}
 	}
 }
@@ -349,12 +382,28 @@ func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	}
 }
 
+func (t *intType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return readInt64Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+		} else {
+			return readInt32Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return readUint64Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+		} else {
+			return readUint32Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+		}
+	}
+}
+
 func (t *dateType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newInt32ColumnIndexer()
 }
 
-func (t *dateType) NewDictionary(bufferSize int) Dictionary {
-	return newInt32Dictionary(t, bufferSize)
+func (t *dateType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newInt32Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -365,6 +414,10 @@ func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newInt32ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *dateType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readInt32Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	if t.Unit.Millis != nil {
 		return newInt32ColumnIndexer()
@@ -373,11 +426,11 @@ func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	}
 }
 
-func (t *timeType) NewDictionary(bufferSize int) Dictionary {
+func (t *timeType) NewDictionary(columnIndex, bufferSize int) Dictionary {
 	if t.Unit.Millis != nil {
-		return newInt32Dictionary(t, bufferSize)
+		return newInt32Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 	} else {
-		return newInt64Dictionary(t, bufferSize)
+		return newInt64Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 	}
 }
 
@@ -397,12 +450,20 @@ func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	}
 }
 
+func (t *timeType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	if t.Unit.Millis != nil {
+		return readInt32Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+	} else {
+		return readInt64Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+	}
+}
+
 func (t *timestampType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newInt64ColumnIndexer()
 }
 
-func (t *timestampType) NewDictionary(bufferSize int) Dictionary {
-	return newInt64Dictionary(t, bufferSize)
+func (t *timestampType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newInt64Dictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -412,3 +473,7 @@ func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffe
 func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newInt64ColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
+
+func (t *timestampType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readInt64Dictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
diff --git a/type_go18.go b/type_go18.go
index 1a229b8..31890d1 100644
--- a/type_go18.go
+++ b/type_go18.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 
 	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
 )
 
@@ -44,8 +45,8 @@ 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]) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, t.class)
 }
 
 func (t primitiveType[T]) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -56,6 +57,10 @@ func (t primitiveType[T]) NewColumnReader(columnIndex, bufferSize int) ColumnRea
 	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, t.class)
 }
 
+func (t primitiveType[T]) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, t.class)
+}
+
 type byteArrayType struct{}
 
 func (t byteArrayType) String() string { return "BYTE_ARRAY" }
@@ -82,8 +87,8 @@ func (t byteArrayType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	return newByteArrayColumnIndexer(sizeLimit)
 }
 
-func (t byteArrayType) NewDictionary(bufferSize int) Dictionary {
-	return newByteArrayDictionary(t, bufferSize)
+func (t byteArrayType) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t byteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -94,6 +99,10 @@ func (t byteArrayType) NewColumnReader(columnIndex, bufferSize int) ColumnReader
 	return newByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t byteArrayType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 type fixedLenByteArrayType struct{ length int }
 
 func (t *fixedLenByteArrayType) String() string {
@@ -120,8 +129,8 @@ 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) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
 func (t *fixedLenByteArrayType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -132,6 +141,10 @@ func (t *fixedLenByteArrayType) NewColumnReader(columnIndex, bufferSize int) Col
 	return newFixedLenByteArrayColumnReader(t, makeColumnIndex(columnIndex), bufferSize)
 }
 
+func (t *fixedLenByteArrayType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readFixedLenByteArrayDictionary(t, makeColumnIndex(columnIndex), numValues, decoder)
+}
+
 // FixedLenByteArrayType constructs a type for fixed-length values of the given
 // size (in bytes).
 func FixedLenByteArrayType(length int) Type { return &fixedLenByteArrayType{length: length} }
@@ -152,18 +165,18 @@ func (t *intType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	}
 }
 
-func (t *intType) NewDictionary(bufferSize int) Dictionary {
+func (t *intType) NewDictionary(columnIndex, bufferSize int) Dictionary {
 	if t.IsSigned {
 		if t.BitWidth == 64 {
-			return newDictionary(t, bufferSize, &int64Class)
+			return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
 		} else {
-			return newDictionary(t, bufferSize, &int32Class)
+			return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
 		}
 	} else {
 		if t.BitWidth == 64 {
-			return newDictionary(t, bufferSize, &uint64Class)
+			return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &uint64Class)
 		} else {
-			return newDictionary(t, bufferSize, &uint32Class)
+			return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &uint32Class)
 		}
 	}
 }
@@ -200,12 +213,28 @@ func (t *intType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	}
 }
 
+func (t *intType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	if t.IsSigned {
+		if t.BitWidth == 64 {
+			return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &int64Class)
+		} else {
+			return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &int32Class)
+		}
+	} else {
+		if t.BitWidth == 64 {
+			return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &uint64Class)
+		} else {
+			return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &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) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
 }
 
 func (t *dateType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -216,6 +245,10 @@ func (t *dateType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
 }
 
+func (t *dateType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &int64Class)
+}
+
 func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	if t.Unit.Millis != nil {
 		return newColumnIndexer(&int32Class)
@@ -224,11 +257,11 @@ func (t *timeType) NewColumnIndexer(sizeLimit int) ColumnIndexer {
 	}
 }
 
-func (t *timeType) NewDictionary(bufferSize int) Dictionary {
+func (t *timeType) NewDictionary(columnIndex, bufferSize int) Dictionary {
 	if t.Unit.Millis != nil {
-		return newDictionary(t, bufferSize, &int32Class)
+		return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int32Class)
 	} else {
-		return newDictionary(t, bufferSize, &int64Class)
+		return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
 	}
 }
 
@@ -248,12 +281,20 @@ func (t *timeType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	}
 }
 
+func (t *timeType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	if t.Unit.Millis != nil {
+		return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &int32Class)
+	} else {
+		return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &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) NewDictionary(columnIndex, bufferSize int) Dictionary {
+	return newDictionary(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
 }
 
 func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffer {
@@ -263,3 +304,7 @@ func (t *timestampType) NewColumnBuffer(columnIndex, bufferSize int) ColumnBuffe
 func (t *timestampType) NewColumnReader(columnIndex, bufferSize int) ColumnReader {
 	return newColumnReader(t, makeColumnIndex(columnIndex), bufferSize, &int64Class)
 }
+
+func (t *timestampType) ReadDictionary(columnIndex, numValues int, decoder encoding.Decoder) (Dictionary, error) {
+	return readDictionary(t, makeColumnIndex(columnIndex), numValues, decoder, &int64Class)
+}
diff --git a/writer.go b/writer.go
index 84e9001..afc182e 100644
--- a/writer.go
+++ b/writer.go
@@ -272,9 +272,10 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		encoding, compression := encodingAndCompressionOf(leaf.node)
 		dictionary := Dictionary(nil)
 		columnType := leaf.node.Type()
+		columnIndex := int(leaf.columnIndex)
 
 		if isDictionaryEncoding(encoding) {
-			dictionary = columnType.NewDictionary(0)
+			dictionary = columnType.NewDictionary(columnIndex, defaultDictBufferSize)
 			columnType = dictionary.Type()
 		}
 
@@ -765,7 +766,7 @@ func (c *writerColumn) flushFilterPages() error {
 
 		// If there is a dictionary, we need to only write the dictionary.
 		if dict := c.dictionary; dict != nil {
-			return dict.WriteTo(c.page.filter)
+			return dict.Page().WriteTo(c.page.filter)
 		}
 
 		for _, page := range c.filter {
@@ -1125,7 +1126,7 @@ func (c *writerColumn) writeDictionaryPage(output io.Writer, dict Dictionary) er
 	c.page.uncompressed.Reset(p)
 	c.dict.encoder.Reset(&c.page.uncompressed)
 
-	if err := dict.WriteTo(&c.dict.encoder); err != nil {
+	if err := dict.Page().WriteTo(&c.dict.encoder); err != nil {
 		return fmt.Errorf("writing parquet dictionary page: %w", err)
 	}
 	if err := p.Close(); err != nil {


Test patch:
diff --git a/buffer_test.go b/buffer_test.go
index b5745c9..d467b94 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -160,7 +160,7 @@ func TestBuffer(t *testing.T) {
 				typ      parquet.Type
 			}{
 				{scenario: "plain", typ: test.typ},
-				{scenario: "indexed", typ: test.typ.NewDictionary(0).Type()},
+				{scenario: "indexed", typ: test.typ.NewDictionary(0, 0).Type()},
 			} {
 				t.Run(config.scenario, func(t *testing.T) {
 					for _, mod := range [...]struct {
diff --git a/column_mapping_test.go b/column_mapping_test.go
new file mode 100644
index 0000000..a19214c
--- /dev/null
+++ b/column_mapping_test.go
@@ -0,0 +1,29 @@
+package parquet_test
+
+import (
+	"fmt"
+
+	"github.com/segmentio/parquet-go"
+)
+
+func ExampleColumnMapping() {
+	schema := parquet.SchemaOf(struct {
+		FirstName  string `parquet:"first_name"`
+		LastName   string `parquet:"last_name"`
+		Attributes []struct {
+			Name  string `parquet:"name"`
+			Value string `parquet:"value"`
+		} `parquet:"attributes"`
+	}{})
+
+	mapping := parquet.ColumnMappingOf(schema)
+	fmt.Println(mapping)
+
+	// Output:
+	// {
+	//    0 => "attributes.name"
+	//    1 => "attributes.value"
+	//    2 => "first_name"
+	//    3 => "last_name"
+	// }
+}
diff --git a/dictionary_test.go b/dictionary_test.go
index 2ceb707..67fd023 100644
--- a/dictionary_test.go
+++ b/dictionary_test.go
@@ -42,7 +42,7 @@ func benchmarkDictionary(b *testing.B, do func(*testing.B, parquet.Type)) {
 
 func randDictionaryPage(typ parquet.Type, values []parquet.Value) parquet.BufferedPage {
 	const bufferSize = 64 * 1024
-	dict := typ.NewDictionary(4 * bufferSize)
+	dict := typ.NewDictionary(0, 4*bufferSize)
 	buf := dict.Type().NewColumnBuffer(0, bufferSize)
 	buf.WriteValues(values)
 	return buf.Page()
diff --git a/page_go18_test.go b/page_go18_test.go
index a06e62a..2b8dc5c 100644
--- a/page_go18_test.go
+++ b/page_go18_test.go
@@ -8,6 +8,7 @@ import (
 	"math/rand"
 	"reflect"
 	"testing"
+	"unsafe"
 
 	"github.com/segmentio/parquet-go"
 	"github.com/segmentio/parquet-go/deprecated"
@@ -50,6 +51,8 @@ func randValue[T plain.Type](r *rand.Rand) (v T) {
 }
 
 func testPageOf[T plain.Type](t *testing.T) {
+	var zero T
+	sizeof := int(unsafe.Sizeof(zero))
 	schema := parquet.SchemaOf(struct{ Value T }{})
 	r := rand.New(rand.NewSource(0))
 
@@ -61,12 +64,12 @@ func testPageOf[T plain.Type](t *testing.T) {
 					1: randValue[T](r),
 				}
 				n, err := w.(io.Writer).Write(cast.SliceToBytes(values))
-				return values[:n], err
+				return values[:n/sizeof], err
 			},
 			read: func(r parquet.ValueReader) ([]T, error) {
 				values := make([]T, 2)
 				n, err := r.(io.Reader).Read(cast.SliceToBytes(values))
-				return values[:n], err
+				return values[:n/sizeof], err
 			},
 		})
 	})
diff --git a/search_test.go b/search_test.go
new file mode 100644
index 0000000..76f40ac
--- /dev/null
+++ b/search_test.go
@@ -0,0 +1,98 @@
+package parquet_test
+
+import (
+	"fmt"
+	"math"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+)
+
+func assertCompare(t *testing.T, a, b parquet.Value, cmp func(parquet.Value, parquet.Value) int, want int) {
+	if got := cmp(a, b); got != want {
+		t.Errorf("compare(%v, %v): got=%d want=%d", a, b, got, want)
+	}
+}
+
+func TestCompareNullsFirst(t *testing.T) {
+	cmp := parquet.CompareNullsFirst(parquet.Int32Type.Compare)
+	assertCompare(t, parquet.Value{}, parquet.Value{}, cmp, 0)
+	assertCompare(t, parquet.Value{}, parquet.ValueOf(int32(0)), cmp, -1)
+	assertCompare(t, parquet.ValueOf(int32(0)), parquet.Value{}, cmp, +1)
+	assertCompare(t, parquet.ValueOf(int32(0)), parquet.ValueOf(int32(1)), cmp, -1)
+}
+
+func TestCompareNullsLast(t *testing.T) {
+	cmp := parquet.CompareNullsLast(parquet.Int32Type.Compare)
+	assertCompare(t, parquet.Value{}, parquet.Value{}, cmp, 0)
+	assertCompare(t, parquet.Value{}, parquet.ValueOf(int32(0)), cmp, +1)
+	assertCompare(t, parquet.ValueOf(int32(0)), parquet.Value{}, cmp, -1)
+	assertCompare(t, parquet.ValueOf(int32(0)), parquet.ValueOf(int32(1)), cmp, -1)
+}
+
+func TestSearchBinary(t *testing.T) {
+	testSearch(t, [][]int32{
+		{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
+		{10, 10, 10, 10},
+		{21, 22, 23, 24, 25},
+		{30},
+		{31},
+		{32},
+		{42, 43, 44, 45, 46, 47, 48, 49},
+	})
+}
+
+func TestSearchLinear(t *testing.T) {
+	testSearch(t, [][]int32{
+		{10, 10, 10, 10},
+		{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
+		{21, 22, 23, 24, 25},
+		{19, 18, 17, 16, 15, 14, 13, 12, 11},
+		{42, 43, 44, 45, 46, 47, 48, 49},
+	})
+}
+
+func testSearch(t *testing.T, pages [][]int32) {
+	indexer := parquet.Int32Type.NewColumnIndexer(0)
+
+	for _, values := range pages {
+		min := values[0]
+		max := values[0]
+
+		for _, v := range values[1:] {
+			switch {
+			case v < min:
+				min = v
+			case v > max:
+				max = v
+			}
+		}
+
+		indexer.IndexPage(int64(len(values)), 0,
+			parquet.ValueOf(min),
+			parquet.ValueOf(max),
+		)
+	}
+
+	formatIndex := indexer.ColumnIndex()
+	columnIndex := parquet.NewColumnIndex(parquet.Int32, &formatIndex)
+
+	for i, values := range pages {
+		t.Run(fmt.Sprintf("page#%02d", i), func(t *testing.T) {
+			for _, value := range values {
+				v := parquet.ValueOf(value)
+				j := parquet.Search(columnIndex, v, parquet.Int32Type)
+
+				if i != j {
+					t.Errorf("searching for value %v: got=%d want=%d", v, j, i)
+				}
+			}
+
+			for _, test := range []int32{math.MinInt32, math.MaxInt32} {
+				if page := parquet.Search(columnIndex, parquet.ValueOf(test), parquet.Int32Type); page != len(pages) {
+					t.Errorf("search for non-existing value %v: got=%d want=%d", test, page, len(pages))
+				}
+			}
+		})
+	}
+}