Repository: segmentio/parquet-go
Pull request #205 (2022-06-02 15:58:14)

Optimize writing optional columns (part 1)
This PR is the first part of a refactor intended to optimize the encoding of optional columns.

I modified the `ColumnBuffer` and `Dictionary` interface by adding an unexported method which replaces the type switch dispatch we were previously doing, and both enables the optimizations on the optional column and greatly reduced code duplication; @kevinburkesegment I believe this should address the comment you made in https://github.com/segmentio/parquet-go/pull/186#discussion_r880851224

```
name                                          old time/op  new time/op  delta
GenericBuffer/benchmarkRowType/go1.18          751ns ± 0%   751ns ± 0%      ~     (p=0.424 n=10+10)
GenericBuffer/booleanColumn/go1.18             316ns ± 0%   312ns ± 0%    -1.06%  (p=0.000 n=8+9)
GenericBuffer/int32Column/go1.18               236ns ± 0%   230ns ± 0%    -2.64%  (p=0.000 n=10+9)
GenericBuffer/int64Column/go1.18               308ns ± 0%   304ns ± 0%    -1.32%  (p=0.000 n=9+10)
GenericBuffer/floatColumn/go1.18               236ns ± 0%   230ns ± 0%    -2.61%  (p=0.000 n=9+10)
GenericBuffer/doubleColumn/go1.18              308ns ± 0%   304ns ± 0%    -1.29%  (p=0.000 n=9+10)
GenericBuffer/byteArrayColumn/go1.18          9.90µs ± 1%  9.91µs ± 1%      ~     (p=0.400 n=9+10)
GenericBuffer/fixedLenByteArrayColumn/go1.18  5.61µs ± 4%  5.45µs ± 0%    -2.86%  (p=0.000 n=8+10)
GenericBuffer/stringColumn/go1.18             6.21µs ± 0%  6.27µs ± 0%    +0.99%  (p=0.000 n=10+10)
GenericBuffer/indexedStringColumn/go1.18      40.2µs ± 1%  29.2µs ± 0%   -27.45%  (p=0.000 n=9+8)
GenericBuffer/uuidColumn/go1.18                392ns ± 0%   391ns ± 1%      ~     (p=0.138 n=10+10)
GenericBuffer/mapColumn/go1.18                 701µs ± 0%   622µs ± 1%   -11.36%  (p=0.000 n=10+8)
GenericBuffer/decimalColumn/go1.18             308ns ± 0%   303ns ± 0%    -1.46%  (p=0.000 n=10+9)
GenericBuffer/contact/go1.18                  13.0µs ± 0%  13.0µs ± 0%    +0.33%  (p=0.026 n=8+8)
GenericBuffer/paddedBooleanColumn/go1.18       224ns ± 1%   216ns ± 0%    -3.26%  (p=0.000 n=10+10)
GenericBuffer/optionalInt32Column/go1.18      10.8µs ± 0%   2.0µs ± 0%   -81.77%  (p=0.000 n=10+10)

name                                          old row/s    new row/s    delta
GenericBuffer/benchmarkRowType/go1.18          1.33G ± 0%   1.33G ± 0%      ~     (p=0.481 n=10+10)
GenericBuffer/booleanColumn/go1.18             3.17G ± 0%   3.20G ± 0%    +1.07%  (p=0.000 n=8+9)
GenericBuffer/int32Column/go1.18               4.23G ± 0%   4.35G ± 0%    +2.71%  (p=0.000 n=10+9)
GenericBuffer/int64Column/go1.18               3.25G ± 0%   3.29G ± 0%    +1.33%  (p=0.000 n=9+10)
GenericBuffer/floatColumn/go1.18               4.23G ± 0%   4.34G ± 0%    +2.68%  (p=0.000 n=9+10)
GenericBuffer/doubleColumn/go1.18              3.25G ± 0%   3.29G ± 0%    +1.30%  (p=0.000 n=9+10)
GenericBuffer/byteArrayColumn/go1.18            101M ± 1%    101M ± 1%      ~     (p=0.400 n=9+10)
GenericBuffer/fixedLenByteArrayColumn/go1.18    178M ± 4%    183M ± 0%    +2.93%  (p=0.000 n=8+10)
GenericBuffer/stringColumn/go1.18               161M ± 0%    159M ± 0%    -0.98%  (p=0.000 n=10+10)
GenericBuffer/indexedStringColumn/go1.18       24.9M ± 1%   34.3M ± 0%   +37.83%  (p=0.000 n=9+8)
GenericBuffer/uuidColumn/go1.18                2.55G ± 0%   2.56G ± 1%      ~     (p=0.143 n=10+10)
GenericBuffer/mapColumn/go1.18                 1.43M ± 0%   1.61M ± 1%   +12.82%  (p=0.000 n=10+8)
GenericBuffer/decimalColumn/go1.18             3.25G ± 0%   3.30G ± 0%    +1.48%  (p=0.000 n=10+9)
GenericBuffer/contact/go1.18                   77.2M ± 0%   76.9M ± 0%    -0.33%  (p=0.028 n=8+8)
GenericBuffer/paddedBooleanColumn/go1.18       4.47G ± 1%   4.62G ± 0%    +3.36%  (p=0.000 n=10+10)
GenericBuffer/optionalInt32Column/go1.18       92.6M ± 0%  507.6M ± 0%  +448.49%  (p=0.000 n=10+10)
```


Patch:
diff --git a/bitmap.go b/bitmap.go
new file mode 100644
index 0000000..108cc8b
--- /dev/null
+++ b/bitmap.go
@@ -0,0 +1,43 @@
+package parquet
+
+import "sync"
+
+type bitmap struct {
+	bits []uint64
+}
+
+func (m *bitmap) reset(size int) {
+	size = (size + 63) / 64
+	if cap(m.bits) < size {
+		m.bits = make([]uint64, size, 2*size)
+	} else {
+		m.bits = m.bits[:size]
+		m.clear()
+	}
+}
+
+func (m *bitmap) clear() {
+	for i := range m.bits {
+		m.bits[i] = 0
+	}
+}
+
+var (
+	bitmapPool sync.Pool // *bitmap
+)
+
+func acquireBitmap(n int) *bitmap {
+	b, _ := bitmapPool.Get().(*bitmap)
+	if b == nil {
+		b = &bitmap{bits: make([]uint64, n, 2*n)}
+	} else {
+		b.reset(n)
+	}
+	return b
+}
+
+func releaseBitmap(b *bitmap) {
+	if b != nil {
+		bitmapPool.Put(b)
+	}
+}
diff --git a/buffer_go18.go b/buffer_go18.go
index 6bc959f..5eb2433 100644
--- a/buffer_go18.go
+++ b/buffer_go18.go
@@ -12,9 +12,8 @@ import (
 //
 // See GenericWriter for details about the benefits over the classic Buffer API.
 type GenericBuffer[T any] struct {
-	base    Buffer
-	write   bufferFunc[T]
-	columns columnBufferWriter
+	base  Buffer
+	write bufferFunc[T]
 }
 
 // NewGenericBuffer is like NewBuffer but returns a GenericBuffer[T] suited to write
@@ -43,10 +42,6 @@ func NewGenericBuffer[T any](options ...RowGroupOption) *GenericBuffer[T] {
 	}
 	buf.base.configure(config.Schema)
 	buf.write = bufferFuncOf[T](t, config.Schema)
-	buf.columns = columnBufferWriter{
-		columns: buf.base.columns,
-		values:  make([]Value, 0, defaultValueBufferSize),
-	}
 	return buf
 }
 
@@ -77,8 +72,7 @@ func makeBufferFunc[T any](t reflect.Type, schema *Schema) bufferFunc[T] {
 	size := t.Size()
 	writeRows := writeRowsFuncOf(t, schema, nil)
 	return func(buf *GenericBuffer[T], rows []T) (n int, err error) {
-		defer buf.columns.clear()
-		err = writeRows(&buf.columns, makeArray(rows), size, 0, columnLevels{})
+		err = writeRows(buf.base.columns, makeArray(rows), size, 0, columnLevels{})
 		if err == nil {
 			n = len(rows)
 		}
diff --git a/column_buffer.go b/column_buffer.go
index a4ff06c..c74a507 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -16,6 +16,12 @@ import (
 //
 // ColumnBuffer implements sort.Interface as a way to support reordering the
 // rows that have been written to it.
+//
+// The current implementation has a limitation which prevents applications from
+// providing custom versions of this interface because it contains unexported
+// methods. The only way to create ColumnBuffer values is to call the
+// NewColumnBuffer of Type instances. This limitation may be lifted in future
+// releases.
 type ColumnBuffer interface {
 	// Exposes a read-only view of the column buffer.
 	ColumnChunk
@@ -58,6 +64,19 @@ type ColumnBuffer interface {
 
 	// Returns the size of the column buffer in bytes.
 	Size() int64
+
+	// This method is employed to write rows from arrays of Go values into the
+	// column buffer. The method is currently unexported because it uses unsafe
+	// APIs which would be difficult for applications to leverage, increasing
+	// the risk of introducing bugs in the code. As a consequence, applications
+	// cannot use custom implementations of the ColumnBuffer interface since
+	// they cannot declare an unexported method that would match this signature.
+	// It means that in order to create a ColumnBuffer value, programs need to
+	// go through a call to NewColumnBuffer on a Type instance. We make this
+	// trade off for now as it is preferrable to optimize for safety over
+	// extensibility in the public APIs, we might revisit in the future if we
+	// learn about valid use cases for custom column buffer types.
+	writeValues(rows array, size, offset uintptr, levels columnLevels)
 }
 
 type array struct {
@@ -65,10 +84,33 @@ type array struct {
 	len int
 }
 
+func makeValueArray(values []Value) array {
+	return *(*array)(unsafe.Pointer(&values))
+}
+
 func (a array) index(i int, size, offset uintptr) unsafe.Pointer {
 	return unsafe.Add(a.ptr, uintptr(i)*size+offset)
 }
 
+func (a array) slice(i, j int, size, offset uintptr) array {
+	if i < 0 || i > a.len || j < 0 || j > a.len {
+		panic("slice index out of bounds")
+	}
+	if i > j {
+		panic("negative slice length")
+	}
+	return array{
+		ptr: a.index(i, size, offset),
+		len: j - i,
+	}
+}
+
+type columnLevels struct {
+	repetitionDepth byte
+	repetitionLevel byte
+	definitionLevel byte
+}
+
 func columnIndexOfNullable(base ColumnBuffer, maxDefinitionLevel byte, definitionLevels []byte) ColumnIndex {
 	return &nullableColumnIndex{
 		ColumnIndex:        base.ColumnIndex(),
@@ -127,6 +169,7 @@ func (col *reversedColumnBuffer) Less(i, j int) bool { return col.ColumnBuffer.L
 // column or one of its parent(s) are marked optional.
 type optionalColumnBuffer struct {
 	base               ColumnBuffer
+	reordered          bool
 	maxDefinitionLevel byte
 	rows               []int32
 	sortIndex          []int32
@@ -148,6 +191,7 @@ func newOptionalColumnBuffer(base ColumnBuffer, maxDefinitionLevel byte, nullOrd
 func (col *optionalColumnBuffer) Clone() ColumnBuffer {
 	return &optionalColumnBuffer{
 		base:               col.base.Clone(),
+		reordered:          col.reordered,
 		maxDefinitionLevel: col.maxDefinitionLevel,
 		rows:               append([]int32{}, col.rows...),
 		definitionLevels:   append([]byte{}, col.definitionLevels...),
@@ -188,44 +232,44 @@ 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
+	// 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.
+	if col.reordered {
+		numNulls := countLevelsNotEqual(col.definitionLevels, col.maxDefinitionLevel)
+		numValues := len(col.rows) - numNulls
+
+		if numValues > 0 {
+			if cap(col.sortIndex) < numValues {
+				col.sortIndex = make([]int32, numValues)
+			}
+			sortIndex := col.sortIndex[:numValues]
+			i := 0
+			for _, j := range col.rows {
+				if j >= 0 {
+					sortIndex[j] = int32(i)
+					i++
+				}
+			}
 
-	if numValues > 0 {
-		if cap(col.sortIndex) < numValues {
-			col.sortIndex = make([]int32, numValues)
+			// Cyclic sort: O(N)
+			for i := range sortIndex {
+				for j := int(sortIndex[i]); i != j; j = int(sortIndex[i]) {
+					col.base.Swap(i, j)
+					sortIndex[i], sortIndex[j] = sortIndex[j], sortIndex[i]
+				}
+			}
 		}
-		sortIndex := col.sortIndex[:numValues]
+
 		i := 0
-		for _, j := range col.rows {
-			if j >= 0 {
-				sortIndex[j] = int32(i)
+		for _, r := range col.rows {
+			if r >= 0 {
+				col.rows[i] = int32(i)
 				i++
 			}
 		}
 
-		// Cyclic sort: O(N)
-		for i := range sortIndex {
-			for j := int(sortIndex[i]); i != j; j = int(sortIndex[i]) {
-				col.base.Swap(i, j)
-				sortIndex[i], sortIndex[j] = sortIndex[j], sortIndex[i]
-			}
-		}
-	}
-
-	i := 0
-	for _, r := range col.rows {
-		if r >= 0 {
-			col.rows[i] = int32(i)
-			i++
-		}
+		col.reordered = false
 	}
 
 	return newOptionalPage(col.base.Page(), col.maxDefinitionLevel, col.definitionLevels)
@@ -261,6 +305,7 @@ func (col *optionalColumnBuffer) Swap(i, j int) {
 	// swap its values at indexes i and j. We swap the row indexes only, then
 	// reorder the underlying buffer using a cyclic sort when the buffer is
 	// materialized into a page view.
+	col.reordered = true
 	col.rows[i], col.rows[j] = col.rows[j], col.rows[i]
 	col.definitionLevels[i], col.definitionLevels[j] = col.definitionLevels[j], col.definitionLevels[i]
 }
@@ -308,10 +353,46 @@ func (col *optionalColumnBuffer) WriteValues(values []Value) (n int, err error)
 			}
 		}
 	}
-
 	return n, nil
 }
 
+func (col *optionalColumnBuffer) writeValues(rows array, size, offset uintptr, levels columnLevels) {
+	// The row count is zero when writing an null optional value, in which case
+	// we still need to output a row to the buffer to record the definition
+	// level.
+	if rows.len == 0 {
+		col.definitionLevels = append(col.definitionLevels, 0)
+		col.rows = append(col.rows, -1)
+		return
+	}
+
+	col.definitionLevels = appendLevel(col.definitionLevels, levels.definitionLevel, rows.len)
+
+	i := len(col.rows)
+	j := len(col.rows) + rows.len
+
+	if j <= cap(col.rows) {
+		col.rows = col.rows[:j]
+	} else {
+		tmp := make([]int32, j, 2*j)
+		copy(tmp, col.rows)
+		col.rows = tmp
+	}
+
+	newRows := col.rows[i:]
+	if levels.definitionLevel != col.maxDefinitionLevel {
+		for i := range newRows {
+			newRows[i] = -1
+		}
+	} else {
+		rowIndex := int32(col.base.Len())
+		for i := range newRows {
+			newRows[i] = rowIndex + int32(i)
+		}
+		col.base.writeValues(rows, size, offset, levels)
+	}
+}
+
 func (col *optionalColumnBuffer) ReadValuesAt(values []Value, offset int64) (int, error) {
 	length := int64(len(col.definitionLevels))
 	if offset < 0 {
@@ -369,6 +450,7 @@ func (col *optionalColumnBuffer) ReadValuesAt(values []Value, offset int64) (int
 // are marked repeated.
 type repeatedColumnBuffer struct {
 	base               ColumnBuffer
+	reordered          bool
 	maxRepetitionLevel byte
 	maxDefinitionLevel byte
 	rows               []region
@@ -405,6 +487,7 @@ func newRepeatedColumnBuffer(base ColumnBuffer, maxRepetitionLevel, maxDefinitio
 func (col *repeatedColumnBuffer) Clone() ColumnBuffer {
 	return &repeatedColumnBuffer{
 		base:               col.base.Clone(),
+		reordered:          col.reordered,
 		maxRepetitionLevel: col.maxRepetitionLevel,
 		maxDefinitionLevel: col.maxDefinitionLevel,
 		rows:               append([]region{}, col.rows...),
@@ -447,7 +530,7 @@ func (col *repeatedColumnBuffer) Pages() Pages {
 }
 
 func (col *repeatedColumnBuffer) Page() BufferedPage {
-	if repeatedRowsHaveBeenReordered(col.rows) {
+	if col.reordered {
 		if col.reordering == nil {
 			col.reordering = col.Clone().(*repeatedColumnBuffer)
 		}
@@ -496,6 +579,7 @@ func (col *repeatedColumnBuffer) Page() BufferedPage {
 		}
 
 		col.swapReorderingBuffer(column)
+		col.reordered = false
 	}
 
 	return newRepeatedPage(
@@ -559,6 +643,7 @@ func (col *repeatedColumnBuffer) Swap(i, j int) {
 	// column buffer when its view is materialized into a page by creating a
 	// copy and writing rows back to it following the order of rows in the
 	// repeated column buffer.
+	col.reordered = true
 	col.rows[i], col.rows[j] = col.rows[j], col.rows[i]
 }
 
@@ -632,40 +717,31 @@ func (col *repeatedColumnBuffer) writeRow(row []Value) error {
 	return nil
 }
 
-func (col *repeatedColumnBuffer) ReadValuesAt(values []Value, offset int64) (int, error) {
-	// TODO:
-	panic("NOT IMPLEMENTED")
-}
+func (col *repeatedColumnBuffer) writeValues(row array, size, offset uintptr, levels columnLevels) {
+	if levels.repetitionLevel == 0 {
+		col.rows = append(col.rows, region{
+			offset:     uint32(len(col.repetitionLevels)),
+			baseOffset: uint32(col.base.NumValues()),
+		})
+	}
 
-func optionalRowsHaveBeenReordered(rows []int32) bool {
-	i := int32(0)
-	for _, row := range rows {
-		if row < 0 {
-			// Skip any row that is null.
-			continue
-		}
+	if row.len == 0 {
+		col.repetitionLevels = append(col.repetitionLevels, levels.repetitionLevel)
+		col.definitionLevels = append(col.definitionLevels, levels.definitionLevel)
+		return
+	}
 
-		// If rows have been reordered the indices are not increasing exactly
-		// one by one.
-		if row != i {
-			return true
-		}
+	col.repetitionLevels = appendLevel(col.repetitionLevels, levels.repetitionLevel, row.len)
+	col.definitionLevels = appendLevel(col.definitionLevels, levels.definitionLevel, row.len)
 
-		// Only increment the index if the row is not null.
-		i++
+	if levels.definitionLevel == col.maxDefinitionLevel {
+		col.base.writeValues(row, size, offset, levels)
 	}
-	return false
 }
 
-func repeatedRowsHaveBeenReordered(rows []region) bool {
-	lastOffset := uint32(0)
-	for _, row := range rows {
-		if row.offset < lastOffset {
-			return true
-		}
-		lastOffset = row.offset
-	}
-	return false
+func (col *repeatedColumnBuffer) ReadValuesAt(values []Value, offset int64) (int, error) {
+	// TODO:
+	panic("NOT IMPLEMENTED")
 }
 
 // =============================================================================
@@ -765,21 +841,17 @@ func (col *booleanColumnBuffer) WriteBooleans(values []bool) (int, error) {
 		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
 		len: len(values),
 	}
-	col.writeValues(rows, unsafe.Sizeof(false), 0)
+	col.writeValues(rows, unsafe.Sizeof(false), 0, columnLevels{})
 	return len(values), nil
 }
 
 func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *booleanColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *booleanColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	numBytes := bits.ByteCount(uint(col.numValues) + uint(rows.len))
 	if cap(col.bits) < numBytes {
 		col.bits = append(make([]byte, 0, 2*cap(col.bits)), col.bits...)
@@ -915,16 +987,12 @@ func (col *int32ColumnBuffer) WriteInt32s(values []int32) (int, error) {
 }
 
 func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *int32ColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *int32ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]int32, 0, max(n, 2*cap(col.values))), col.values...)
 	}
@@ -1014,16 +1082,12 @@ func (col *int64ColumnBuffer) WriteInt64s(values []int64) (int, error) {
 }
 
 func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *int64ColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *int64ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]int64, 0, max(n, 2*cap(col.values))), col.values...)
 	}
@@ -1119,7 +1183,7 @@ func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
 	return len(values), nil
 }
 
-func (col *int96ColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *int96ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	for i := 0; i < rows.len; i++ {
 		p := rows.index(i, size, offset)
 		col.values = append(col.values, *(*deprecated.Int96)(p))
@@ -1207,16 +1271,12 @@ func (col *floatColumnBuffer) WriteFloats(values []float32) (int, error) {
 }
 
 func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *floatColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *floatColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]float32, 0, max(n, 2*cap(col.values))), col.values...)
 	}
@@ -1306,16 +1366,12 @@ func (col *doubleColumnBuffer) WriteDoubles(values []float64) (int, error) {
 }
 
 func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *doubleColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *doubleColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]float64, 0, max(n, 2*cap(col.values))), col.values...)
 	}
@@ -1456,16 +1512,12 @@ func (col *byteArrayColumnBuffer) writeByteArrays(values []byte) (count, bytes i
 }
 
 func (col *byteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
-	rows := array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.ptr))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.ptr), columnLevels{})
 	return len(values), nil
 }
 
-func (col *byteArrayColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *byteArrayColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	for i := 0; i < rows.len; i++ {
 		p := rows.index(i, size, offset)
 		col.append(*(*string)(p))
@@ -1589,22 +1641,27 @@ func (col *fixedLenByteArrayColumnBuffer) WriteValues(values []Value) (int, erro
 	return len(values), nil
 }
 
-func (col *fixedLenByteArrayColumnBuffer) writeValues(rows array, size, offset uintptr) {
-	for i := 0; i < rows.len; i++ {
-		p := rows.index(i, size, offset)
-		col.data = append(col.data, unsafe.Slice((*byte)(p), col.size)...)
+func (col *fixedLenByteArrayColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
+	n := col.size * rows.len
+	i := len(col.data)
+	j := len(col.data) + n
+
+	if cap(col.data) < j {
+		col.data = append(make([]byte, 0, max(i+n, 2*cap(col.data))), col.data...)
 	}
-}
 
-func (col *fixedLenByteArrayColumnBuffer) writeValuesUint128(rows array, size, offset uintptr) {
-	c := cap(col.data)
-	n := len(col.data) + (16 * rows.len)
-	if c < n {
-		col.data = append(make([]byte, 0, max(n, 2*c)), col.data...)
+	col.data = col.data[:j]
+	newData := col.data[i:]
+
+	switch col.size {
+	case 16:
+		writeValuesUint128(newData, rows, size, offset)
+	default:
+		for i := 0; i < rows.len; i++ {
+			p := rows.index(i, size, offset)
+			copy(newData[i*col.size:], unsafe.Slice((*byte)(p), col.size))
+		}
 	}
-	firstIndex := len(col.data)
-	col.data = col.data[:len(col.data)+(16*rows.len)]
-	writeValuesUint128(col.data[firstIndex:], rows, size, offset)
 }
 
 func (col *fixedLenByteArrayColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1688,16 +1745,12 @@ func (col *uint32ColumnBuffer) WriteUint32s(values []uint32) (int, error) {
 }
 
 func (col *uint32ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *uint32ColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *uint32ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]uint32, 0, max(n, 2*cap(col.values))), col.values...)
 	}
@@ -1787,16 +1840,12 @@ func (col *uint64ColumnBuffer) WriteUint64s(values []uint64) (int, error) {
 }
 
 func (col *uint64ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var rows = array{
-		ptr: *(*unsafe.Pointer)(unsafe.Pointer(&values)),
-		len: len(values),
-	}
 	var value Value
-	col.writeValues(rows, unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	col.writeValues(makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
 	return len(values), nil
 }
 
-func (col *uint64ColumnBuffer) writeValues(rows array, size, offset uintptr) {
+func (col *uint64ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
 	if n := len(col.values) + rows.len; n > cap(col.values) {
 		col.values = append(make([]uint64, 0, max(n, 2*cap(col.values))), col.values...)
 	}
diff --git a/column_buffer_amd64.go b/column_buffer_amd64.go
index 976791d..981bb67 100644
--- a/column_buffer_amd64.go
+++ b/column_buffer_amd64.go
@@ -4,12 +4,8 @@ package parquet
 
 import (
 	"unsafe"
-
-	"golang.org/x/sys/cpu"
 )
 
-var hasAVX2 = cpu.X86.HasAVX2
-
 //go:noescape
 func writeValuesBitpack(values unsafe.Pointer, rows array, size, offset uintptr)
 
diff --git a/column_buffer_amd64.s b/column_buffer_amd64.s
index dbcdd98..671df5f 100644
--- a/column_buffer_amd64.s
+++ b/column_buffer_amd64.s
@@ -32,7 +32,7 @@ init:
     JB loop
 
     VPBROADCASTD size+24(FP), Y0
-    VPMULLD scale8x4<>(SB), Y0, Y0
+    VPMULLD ·scale8x4(SB), Y0, Y0
     VPCMPEQD Y1, Y1, Y1
     VPCMPEQD Y2, Y2, Y2
     VPCMPEQD Y3, Y3, Y3
@@ -122,7 +122,7 @@ TEXT ·writeValues32bits(SB), NOSPLIT, $0-40
     SHLQ $3, DI
 
     VPBROADCASTD size+24(FP), Y0
-    VPMULLD scale8x4<>(SB), Y0, Y0
+    VPMULLD ·scale8x4(SB), Y0, Y0
     VPCMPEQD Y1, Y1, Y1
     VPCMPEQD Y2, Y2, Y2
 loop8x4:
@@ -174,7 +174,7 @@ TEXT ·writeValues64bits(SB), NOSPLIT, $0-40
     SHLQ $2, DI
 
     VPBROADCASTQ size+24(FP), Y0
-    VPMULLD scale4x8<>(SB), Y0, Y0
+    VPMULLD ·scale4x8(SB), Y0, Y0
     VPCMPEQD Y1, Y1, Y1
     VPCMPEQD Y2, Y2, Y2
 loop4x8:
@@ -251,18 +251,18 @@ loop1x16:
 done:
     RET
 
-GLOBL scale8x4<>(SB), RODATA|NOPTR, $32
-DATA scale8x4<>+0(SB)/4,  $0
-DATA scale8x4<>+4(SB)/4,  $1
-DATA scale8x4<>+8(SB)/4,  $2
-DATA scale8x4<>+12(SB)/4, $3
-DATA scale8x4<>+16(SB)/4, $4
-DATA scale8x4<>+20(SB)/4, $5
-DATA scale8x4<>+24(SB)/4, $6
-DATA scale8x4<>+28(SB)/4, $7
-
-GLOBL scale4x8<>(SB), RODATA|NOPTR, $32
-DATA scale4x8<>+0(SB)/8,  $0
-DATA scale4x8<>+8(SB)/8,  $1
-DATA scale4x8<>+16(SB)/8, $2
-DATA scale4x8<>+24(SB)/8, $3
+GLOBL ·scale8x4(SB), RODATA|NOPTR, $32
+DATA ·scale8x4+0(SB)/4,  $0
+DATA ·scale8x4+4(SB)/4,  $1
+DATA ·scale8x4+8(SB)/4,  $2
+DATA ·scale8x4+12(SB)/4, $3
+DATA ·scale8x4+16(SB)/4, $4
+DATA ·scale8x4+20(SB)/4, $5
+DATA ·scale8x4+24(SB)/4, $6
+DATA ·scale8x4+28(SB)/4, $7
+
+GLOBL ·scale4x8(SB), RODATA|NOPTR, $32
+DATA ·scale4x8+0(SB)/8,  $0
+DATA ·scale4x8+8(SB)/8,  $1
+DATA ·scale4x8+16(SB)/8, $2
+DATA ·scale4x8+24(SB)/8, $3
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index 10523a0..c864992 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -3,7 +3,7 @@
 package parquet
 
 import (
-	"bytes"
+	"math/bits"
 	"reflect"
 	"unsafe"
 
@@ -25,228 +25,10 @@ func slice[T any](p unsafe.Pointer, n int) []T {
 	return unsafe.Slice((*T)(p), n)
 }
 
-// nullIndexFunc is the type of functions generated by calling nullIndexFuncOf.
-//
-// The function takes an array of values that it assumes the type of (based on
-// the reflect.Type that it was created from) and returns the index of the first
-// null value, or the length of the array if no values were null.
-//
-// A value is null if it is zero for number types, or if it is nil for a slice
-// type. Struct values are never null.
-type nullIndexFunc func(array) int
-
-func nullIndexByte(a array) int {
-	i := bytes.IndexByte(makeSlice[byte](a), 0)
-	if i < 0 {
-		i = a.len
-	}
-	return i
-}
-
-func nullIndexSlice(a array) int {
-	const size = unsafe.Sizeof(([]byte)(nil))
-	for i := 0; i < a.len; i++ {
-		p := *(*unsafe.Pointer)(a.index(i, size, 0))
-		if p == nil {
-			return i
-		}
-	}
-	return a.len
-}
-
-func nullIndex[T comparable](a array) int {
-	var zero T
-	for i, v := range makeSlice[T](a) {
-		if v == zero {
-			return i
-		}
-	}
-	return a.len
-}
-
-func nullIndexFuncOf(t reflect.Type) nullIndexFunc {
-	switch t {
-	case reflect.TypeOf(deprecated.Int96{}):
-		return nullIndex[deprecated.Int96]
-	}
-
-	switch t.Kind() {
-	case reflect.Bool:
-		return nullIndexByte
-
-	case reflect.Int, reflect.Uint:
-		return nullIndex[int]
-
-	case reflect.Int8, reflect.Uint8:
-		return nullIndexByte
-
-	case reflect.Int16, reflect.Uint16:
-		return nullIndex[int16]
-
-	case reflect.Int32, reflect.Uint32:
-		return nullIndex[int32]
-
-	case reflect.Int64, reflect.Uint64:
-		return nullIndex[int64]
-
-	case reflect.Float32:
-		return nullIndex[float32]
-
-	case reflect.Float64:
-		return nullIndex[float64]
-
-	case reflect.String:
-		return nullIndex[string]
-
-	case reflect.Slice:
-		return nullIndexSlice
-
-	case reflect.Array:
-		if t.Elem().Kind() == reflect.Uint8 {
-			return nullIndexFuncOfArray(t)
-		}
-
-	case reflect.Pointer:
-		return nullIndex[unsafe.Pointer]
-
-	case reflect.Struct:
-		return func(a array) int { return a.len }
-	}
-
-	panic("cannot convert Go values of type " + t.String() + " to parquet value")
-}
-
-func nullIndexFuncOfArray(t reflect.Type) nullIndexFunc {
-	arrayLen := t.Len()
-	return func(a array) int {
-		for i := 0; i < a.len; i++ {
-			p := a.index(i, uintptr(arrayLen), 0)
-			b := slice[byte](p, arrayLen)
-			if bytes.Count(b, []byte{0}) == len(b) {
-				return i
-			}
-		}
-		return a.len
-	}
-}
-
-// nonNullIndexFunc is the type of functions generated by calling
-// nonNullIndexFuncOf.
-//
-// The function takes an array of values that it assumes the type of (based on
-// the reflect.Type that it was created from) and returns the index of the first
-// non-null value, or the length of the array if all values were null.
-//
-// A value is null if it is zero for number types, or if it is nil for a slice
-// type. Struct values are never null.
-type nonNullIndexFunc func(array) int
-
-func nonNullIndexSlice(a array) int {
-	const size = unsafe.Sizeof(([]byte)(nil))
-	for i := 0; i < a.len; i++ {
-		p := *(*unsafe.Pointer)(a.index(i, size, 0))
-		if p != nil {
-			return i
-		}
-	}
-	return a.len
-}
-
-func nonNullIndex[T comparable](a array) int {
-	var zero T
-	for i, v := range makeSlice[T](a) {
-		if v != zero {
-			return i
-		}
-	}
-	return a.len
-}
-
-func nonNullIndexFuncOf(t reflect.Type) nonNullIndexFunc {
-	switch t {
-	case reflect.TypeOf(deprecated.Int96{}):
-		return nonNullIndex[deprecated.Int96]
-	}
-
-	switch t.Kind() {
-	case reflect.Bool:
-		return nonNullIndex[bool]
-
-	case reflect.Int, reflect.Uint:
-		return nonNullIndex[int]
-
-	case reflect.Int8, reflect.Uint8:
-		return nonNullIndex[int8]
-
-	case reflect.Int16, reflect.Uint16:
-		return nonNullIndex[int16]
-
-	case reflect.Int32, reflect.Uint32:
-		return nonNullIndex[int32]
-
-	case reflect.Int64, reflect.Uint64:
-		return nonNullIndex[int64]
-
-	case reflect.Float32:
-		return nonNullIndex[float32]
-
-	case reflect.Float64:
-		return nonNullIndex[float64]
-
-	case reflect.String:
-		return nonNullIndex[string]
-
-	case reflect.Slice:
-		return nonNullIndexSlice
-
-	case reflect.Array:
-		if t.Elem().Kind() == reflect.Uint8 {
-			return nonNullIndexFuncOfArray(t)
-		}
-
-	case reflect.Pointer:
-		return nonNullIndex[unsafe.Pointer]
-
-	case reflect.Struct:
-		return func(array) int { return 0 }
-	}
-
-	panic("cannot convert Go values of type " + t.String() + " to parquet value")
-}
-
-func nonNullIndexFuncOfArray(t reflect.Type) nonNullIndexFunc {
-	arrayLen := t.Len()
-	return func(a array) int {
-		for i := 0; i < a.len; i++ {
-			p := a.index(i, uintptr(arrayLen), 0)
-			b := slice[byte](p, arrayLen)
-			if bytes.Count(b, []byte{0}) != len(b) {
-				return i
-			}
-		}
-		return a.len
-	}
-}
-
-type columnLevels struct {
-	columnIndex     int16
-	repetitionDepth byte
-	repetitionLevel byte
-	definitionLevel byte
-}
-
-// columnBufferWriter is a type used to writes rows to column buffers.
-type columnBufferWriter struct {
-	columns []ColumnBuffer // column buffers where row values are written
-	values  []Value        // buffer used to temporarily hold values
-	maxLen  int            // max number of values written to the temp buffer
-}
-
 // writeRowsFunc is the type of functions that apply rows to a set of column
 // buffers.
 //
-// - w is the columnBufferWriter holding the column buffers where the rows are
-//   written.
+// - columns is the array of column buffer where the rows are written.
 //
 // - rows is the array of Go values to write to the column buffers.
 //
@@ -258,7 +40,7 @@ type columnBufferWriter struct {
 // - levels is used to track the column index, repetition and definition levels
 //   of values when writing optional or repeated columns.
 //
-type writeRowsFunc func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error
+type writeRowsFunc func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error
 
 // writeRowsFuncOf generates a writeRowsFunc function for the given Go type and
 // parquet schema. The column path indicates the column that the function is
@@ -266,47 +48,32 @@ type writeRowsFunc func(w *columnBufferWriter, rows array, size, offset uintptr,
 func writeRowsFuncOf(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	switch t {
 	case reflect.TypeOf(deprecated.Int96{}):
-		return (*columnBufferWriter).writeRowsInt96
+		return writeRowsFuncOfRequired(t, schema, path)
 	}
 
 	switch t.Kind() {
-	case reflect.Bool:
-		return (*columnBufferWriter).writeRowsBool
-
-	case reflect.Int, reflect.Uint:
-		return (*columnBufferWriter).writeRowsInt
-
-	case reflect.Int8, reflect.Uint8:
-		return (*columnBufferWriter).writeRowsInt8
-
-	case reflect.Int16, reflect.Uint16:
-		return (*columnBufferWriter).writeRowsInt16
-
-	case reflect.Int32, reflect.Uint32:
-		return (*columnBufferWriter).writeRowsInt32
-
-	case reflect.Int64, reflect.Uint64:
-		return (*columnBufferWriter).writeRowsInt64
-
-	case reflect.Float32:
-		return (*columnBufferWriter).writeRowsFloat32
-
-	case reflect.Float64:
-		return (*columnBufferWriter).writeRowsFloat64
-
-	case reflect.String:
-		return (*columnBufferWriter).writeRowsString
+	case reflect.Bool,
+		reflect.Int,
+		reflect.Uint,
+		reflect.Int32,
+		reflect.Uint32,
+		reflect.Int64,
+		reflect.Uint64,
+		reflect.Float32,
+		reflect.Float64,
+		reflect.String:
+		return writeRowsFuncOfRequired(t, schema, path)
 
 	case reflect.Slice:
 		if t.Elem().Kind() == reflect.Uint8 {
-			return (*columnBufferWriter).writeRowsString
+			return writeRowsFuncOfRequired(t, schema, path)
 		} else {
 			return writeRowsFuncOfSlice(t, schema, path)
 		}
 
 	case reflect.Array:
 		if t.Elem().Kind() == reflect.Uint8 {
-			return writeRowsFuncOfArray(t, schema, path)
+			return writeRowsFuncOfRequired(t, schema, path)
 		}
 
 	case reflect.Pointer:
@@ -322,27 +89,30 @@ func writeRowsFuncOf(t reflect.Type, schema *Schema, path columnPath) writeRowsF
 	panic("cannot convert Go values of type " + t.String() + " to parquet value")
 }
 
-func writeRowsFuncOfArray(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
-	len := t.Len()
-	if len == 16 {
-		return (*columnBufferWriter).writeRowsUUID
-	}
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
-		return w.writeRowsArray(rows, size, offset, levels, len)
+func writeRowsFuncOfRequired(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
+	column := schema.mapping.lookup(path)
+	columnIndex := column.columnIndex
+	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
+		columns[columnIndex].writeValues(rows, size, offset, levels)
+		return nil
 	}
 }
 
 func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, writeRows writeRowsFunc) writeRowsFunc {
-	nullIndex, nonNullIndex := nullIndexFuncOf(t), nonNullIndexFuncOf(t)
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
+	nullIndex := nullIndexFuncOf(t)
+	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
 		if rows.len == 0 {
-			return writeRows(w, rows, size, 0, levels)
+			return writeRows(columns, rows, size, 0, levels)
 		}
 
-		nonNullLevels := levels
-		nonNullLevels.definitionLevel++
+		nulls := acquireBitmap(rows.len)
+		defer releaseBitmap(nulls)
+		nullIndex(nulls.bits, rows, size, offset)
+
+		nullLevels := levels
+		levels.definitionLevel++
 		// In this function, we are dealing with optional values which are
-		// neither pointers nor slices; for example, a []int32 field marked
+		// neither pointers nor slices; for example, a int32 field marked
 		// "optional" in its parent struct.
 		//
 		// We need to find zero values, which should be represented as nulls
@@ -358,30 +128,71 @@ func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, wr
 		// sequences of single values, we do not expect this condition to be a
 		// common case.
 		for i := 0; i < rows.len; {
-			a := array{}
-			p := rows.index(i, size, 0)
-			j := i + nonNullIndex(array{ptr: p, len: rows.len - i})
+			j := 0
+			x := i / 64
+			y := i % 64
+
+			if y != 0 {
+				if b := nulls.bits[x] >> uint(y); b == 0 {
+					x++
+					y = 0
+				} else {
+					y += bits.TrailingZeros64(b)
+					goto writeNulls
+				}
+			}
+
+			for x < len(nulls.bits) && nulls.bits[x] == 0 {
+				x++
+			}
+
+			if x < len(nulls.bits) {
+				y = bits.TrailingZeros64(nulls.bits[x]) % 64
+			}
+
+		writeNulls:
+			if j = x*64 + y; j > rows.len {
+				j = rows.len
+			}
 
 			if i < j {
-				a.ptr = p
-				a.len = j - i
-				if err := writeRows(w, a, size, 0, levels); err != nil {
+				slice := rows.slice(i, j, size, offset)
+				if err := writeRows(columns, slice, size, offset, nullLevels); err != nil {
 					return err
 				}
+				i = j
 			}
 
-			if j < rows.len {
-				p = rows.index(j, size, 0)
-				i = j
-				j = j + nullIndex(array{ptr: p, len: rows.len - j})
-				a.ptr = p
-				a.len = j - i
-				if err := writeRows(w, a, size, 0, nonNullLevels); err != nil {
-					return err
+			if y != 0 {
+				if b := nulls.bits[x] >> uint(y); b == (1<<uint64(y))-1 {
+					x++
+					y = 0
+				} else {
+					y += bits.TrailingZeros64(^b)
+					goto writeNonNulls
 				}
 			}
 
-			i = j
+			for x < len(nulls.bits) && nulls.bits[x] == ^uint64(0) {
+				x++
+			}
+
+			if x < len(nulls.bits) {
+				y = bits.TrailingZeros64(^nulls.bits[x]) % 64
+			}
+
+		writeNonNulls:
+			if j = x*64 + y; j > rows.len {
+				j = rows.len
+			}
+
+			if i < j {
+				slice := rows.slice(i, j, size, offset)
+				if err := writeRows(columns, slice, size, offset, levels); err != nil {
+					return err
+				}
+				i = j
+			}
 		}
 
 		return nil
@@ -398,9 +209,9 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 		// type. In this case, we do not need to increase the definition level
 		// since we are not deailng with an optional field but a pointer to the
 		// row type.
-		return func(w *columnBufferWriter, rows array, size, _ uintptr, levels columnLevels) error {
+		return func(columns []ColumnBuffer, rows array, size, _ uintptr, levels columnLevels) error {
 			if rows.len == 0 {
-				return writeRows(w, rows, size, 0, levels)
+				return writeRows(columns, rows, size, 0, levels)
 			}
 
 			for i := 0; i < rows.len; i++ {
@@ -410,7 +221,7 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 					a.ptr = p
 					a.len = 1
 				}
-				if err := writeRows(w, a, elemSize, 0, levels); err != nil {
+				if err := writeRows(columns, a, elemSize, 0, levels); err != nil {
 					return err
 				}
 			}
@@ -419,9 +230,9 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 		}
 	}
 
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
+	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
 		if rows.len == 0 {
-			return writeRows(w, rows, size, 0, levels)
+			return writeRows(columns, rows, size, 0, levels)
 		}
 
 		for i := 0; i < rows.len; i++ {
@@ -433,7 +244,7 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 				a.len = 1
 				elemLevels.definitionLevel++
 			}
-			if err := writeRows(w, a, elemSize, 0, elemLevels); err != nil {
+			if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
 				return err
 			}
 		}
@@ -446,9 +257,9 @@ func writeRowsFuncOfSlice(t reflect.Type, schema *Schema, path columnPath) write
 	elemType := t.Elem()
 	elemSize := elemType.Size()
 	writeRows := writeRowsFuncOf(elemType, schema, path)
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
+	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
 		if rows.len == 0 {
-			return writeRows(w, rows, size, 0, levels)
+			return writeRows(columns, rows, size, 0, levels)
 		}
 
 		levels.repetitionDepth++
@@ -464,7 +275,7 @@ func writeRowsFuncOfSlice(t reflect.Type, schema *Schema, path columnPath) write
 				elemLevels.definitionLevel++
 			}
 
-			if err := writeRows(w, a, elemSize, 0, elemLevels); err != nil {
+			if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
 				return err
 			}
 
@@ -473,7 +284,7 @@ func writeRowsFuncOfSlice(t reflect.Type, schema *Schema, path columnPath) write
 				a.ptr = a.index(1, elemSize, 0)
 				a.len = n - 1
 
-				if err := writeRows(w, a, elemSize, 0, elemLevels); err != nil {
+				if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
 					return err
 				}
 			}
@@ -485,10 +296,8 @@ func writeRowsFuncOfSlice(t reflect.Type, schema *Schema, path columnPath) write
 
 func writeRowsFuncOfStruct(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	type column struct {
-		columnIndex int16
-		optional    bool
-		offset      uintptr
-		writeRows   writeRowsFunc
+		offset    uintptr
+		writeRows writeRowsFunc
 	}
 
 	fields := structFieldsOf(t)
@@ -515,18 +324,15 @@ func writeRowsFuncOfStruct(t reflect.Type, schema *Schema, path columnPath) writ
 			}
 		}
 
-		columnInfo := schema.mapping.lookup(columnPath)
 		columns[i] = column{
-			columnIndex: columnInfo.columnIndex,
-			offset:      f.Offset,
-			writeRows:   writeRows,
+			offset:    f.Offset,
+			writeRows: writeRows,
 		}
 	}
 
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
+	return func(buffers []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
 		for _, column := range columns {
-			levels.columnIndex = column.columnIndex
-			if err := column.writeRows(w, rows, size, offset+column.offset, levels); err != nil {
+			if err := column.writeRows(buffers, rows, size, offset+column.offset, levels); err != nil {
 				return err
 			}
 		}
@@ -539,29 +345,25 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 	keyType := t.Key()
 	keySize := keyType.Size()
 	writeKeys := writeRowsFuncOf(keyType, schema, keyPath)
-	keyColumnIndex := schema.mapping.lookup(keyPath).columnIndex
 
 	valuePath := path.append("key_value", "value")
 	valueType := t.Elem()
 	valueSize := valueType.Size()
 	writeValues := writeRowsFuncOf(valueType, schema, valuePath)
-	valueColumnIndex := schema.mapping.lookup(valuePath).columnIndex
 
-	writeKeyValues := func(w *columnBufferWriter, keys, values array, levels columnLevels) error {
-		levels.columnIndex = keyColumnIndex
-		if err := writeKeys(w, keys, keySize, 0, levels); err != nil {
+	writeKeyValues := func(columns []ColumnBuffer, keys, values array, levels columnLevels) error {
+		if err := writeKeys(columns, keys, keySize, 0, levels); err != nil {
 			return err
 		}
-		levels.columnIndex = valueColumnIndex
-		if err := writeValues(w, values, valueSize, 0, levels); err != nil {
+		if err := writeValues(columns, values, valueSize, 0, levels); err != nil {
 			return err
 		}
 		return nil
 	}
 
-	return func(w *columnBufferWriter, rows array, size, offset uintptr, levels columnLevels) error {
+	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
 		if rows.len == 0 {
-			return writeKeyValues(w, rows, rows, levels)
+			return writeKeyValues(columns, rows, rows, levels)
 		}
 
 		levels.repetitionDepth++
@@ -572,7 +374,7 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 			m := reflect.NewAt(t, rows.index(i, size, offset)).Elem()
 
 			if m.Len() == 0 {
-				if err := writeKeyValues(w, array{}, array{}, levels); err != nil {
+				if err := writeKeyValues(columns, array{}, array{}, levels); err != nil {
 					return err
 				}
 			} else {
@@ -586,7 +388,7 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 					k := array{ptr: addressOf(mapKey), len: 1}
 					v := array{ptr: addressOf(mapValue), len: 1}
 
-					if err := writeKeyValues(w, k, v, elemLevels); err != nil {
+					if err := writeKeyValues(columns, k, v, elemLevels); err != nil {
 						return err
 					}
 
@@ -602,311 +404,3 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 func addressOf(v reflect.Value) unsafe.Pointer {
 	return (*[2]unsafe.Pointer)(unsafe.Pointer(&v))[1]
 }
-
-func (w *columnBufferWriter) writeRowsBool(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *booleanColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueBoolean(*(*bool)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int64ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	case *uint64ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt64(int64(*(*int)(p)))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt8(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	case *uint32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt32(int32(*(*int8)(p)))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt16(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	case *uint32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt32(int32(*(*int16)(p)))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt32(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	case *uint32ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt32(*(*int32)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt64(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int64ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	case *uint64ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt64(*(*int64)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsInt96(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *int96ColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueInt96(*(*deprecated.Int96)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsFloat32(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *floatColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueFloat(*(*float32)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsFloat64(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *doubleColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueDouble(*(*float64)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsString(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *byteArrayColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueString(ByteArray, *(*string)(p))
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsUUID(rows array, size, offset uintptr, levels columnLevels) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *fixedLenByteArrayColumnBuffer:
-		c.writeValuesUint128(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueByteArray(FixedLenByteArray, (*byte)(p), 16)
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsArray(rows array, size, offset uintptr, levels columnLevels, len int) (err error) {
-	if rows.len == 0 {
-		return w.writeRowsNull(levels)
-	}
-
-	switch c := w.columns[levels.columnIndex].(type) {
-	case *fixedLenByteArrayColumnBuffer:
-		c.writeValues(rows, size, offset)
-	default:
-		w.reset()
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			v := makeValueByteArray(FixedLenByteArray, (*byte)(p), len)
-			v.repetitionLevel = levels.repetitionLevel
-			v.definitionLevel = levels.definitionLevel
-			w.values = append(w.values, v)
-		}
-		_, err = w.columns[levels.columnIndex].WriteValues(w.values)
-	}
-
-	return err
-}
-
-func (w *columnBufferWriter) writeRowsNull(levels columnLevels) error {
-	w.reset()
-	w.values = append(w.values[:0], Value{
-		repetitionLevel: levels.repetitionLevel,
-		definitionLevel: levels.definitionLevel,
-	})
-	_, err := w.columns[levels.columnIndex].WriteValues(w.values)
-	return err
-}
-
-func (w *columnBufferWriter) reset() {
-	if len(w.values) > w.maxLen {
-		w.maxLen = len(w.values)
-	}
-	w.values = w.values[:0]
-}
-
-func (w *columnBufferWriter) clear() {
-	clearValues(w.values[:w.maxLen])
-	w.maxLen = 0
-}
diff --git a/dictionary.go b/dictionary.go
index fd852f5..ab6d0fc 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -3,6 +3,7 @@ package parquet
 import (
 	"bytes"
 	"io"
+	"unsafe"
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
@@ -21,6 +22,12 @@ const (
 //
 // Programs can instantiate dictionaries by call the NewDictionary method of a
 // Type object.
+//
+// The current implementation has a limitation which prevents applications from
+// providing custom versions of this interface because it contains unexported
+// methods. The only way to create Dictionary values is to call the
+// NewDictionary of Type instances. This limitation may be lifted in future
+// releases.
 type Dictionary interface {
 	// Returns the type that the dictionary was created from.
 	Type() Type
@@ -56,6 +63,10 @@ type Dictionary interface {
 	// 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
+
+	// See ColumnBuffer.writeValues for details on the use of unexported methods
+	// on interfaces.
+	insert(indexes []int32, rows array, size, offset uintptr)
 }
 
 // The boolean dictionary always contains two values for true and false.
@@ -82,7 +93,12 @@ func (d *booleanDictionary) Len() int { return int(d.numValues) }
 func (d *booleanDictionary) Index(i int32) Value { return makeValueBoolean(d.valueAt(int(i))) }
 
 func (d *booleanDictionary) Insert(indexes []int32, values []Value) {
-	_ = indexes[:len(values)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *booleanDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[bool]int32, cap(d.bits))
@@ -91,8 +107,8 @@ func (d *booleanDictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Boolean()
+	for i := 0; i < rows.len; i++ {
+		value := *(*bool)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -167,7 +183,12 @@ 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)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *int32Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[int32]int32, cap(d.values))
@@ -176,8 +197,8 @@ func (d *int32Dictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Int32()
+	for i := 0; i < rows.len; i++ {
+		value := *(*int32)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -248,7 +269,12 @@ 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)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *int64Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[int64]int32, cap(d.values))
@@ -257,8 +283,8 @@ func (d *int64Dictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Int64()
+	for i := 0; i < rows.len; i++ {
+		value := *(*int64)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -329,7 +355,19 @@ 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)]
+	d.insertValues(indexes, len(values), func(i int) deprecated.Int96 {
+		return values[i].Int96()
+	})
+}
+
+func (d *int96Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	d.insertValues(indexes, rows.len, func(i int) deprecated.Int96 {
+		return *(*deprecated.Int96)(rows.index(i, size, offset))
+	})
+}
+
+func (d *int96Dictionary) insertValues(indexes []int32, count int, valueAt func(int) deprecated.Int96) {
+	_ = indexes[:count]
 
 	if d.index == nil {
 		d.index = make(map[deprecated.Int96]int32, cap(d.values))
@@ -338,8 +376,8 @@ func (d *int96Dictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Int96()
+	for i := 0; i < count; i++ {
+		value := valueAt(i)
 
 		index, exists := d.index[value]
 		if !exists {
@@ -410,7 +448,12 @@ 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)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *floatDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[float32]int32, cap(d.values))
@@ -419,8 +462,8 @@ func (d *floatDictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Float()
+	for i := 0; i < rows.len; i++ {
+		value := *(*float32)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -491,7 +534,12 @@ 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)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *doubleDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[float64]int32, cap(d.values))
@@ -500,8 +548,8 @@ func (d *doubleDictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Double()
+	for i := 0; i < rows.len; i++ {
+		value := *(*float64)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -586,7 +634,12 @@ func (d *byteArrayDictionary) Index(i int32) Value {
 }
 
 func (d *byteArrayDictionary) Insert(indexes []int32, values []Value) {
-	_ = indexes[:len(values)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.ptr))
+}
+
+func (d *byteArrayDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[string]int32, cap(d.offsets))
@@ -595,27 +648,26 @@ func (d *byteArrayDictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.ByteArray()
+	for i := 0; i < rows.len; i++ {
+		value := *(*string)(rows.index(i, size, offset))
 
-		index, exists := d.index[string(value)]
+		index, exists := d.index[value]
 		if !exists {
 			index = int32(len(d.offsets))
 			value = d.append(value)
-			stringValue := bits.BytesToString(value)
-			d.index[stringValue] = index
+			d.index[value] = index
 		}
 
 		indexes[i] = index
 	}
 }
 
-func (d *byteArrayDictionary) append(value []byte) []byte {
+func (d *byteArrayDictionary) append(value string) string {
 	offset := len(d.values)
-	d.values = plain.AppendByteArray(d.values, value)
+	d.values = plain.AppendByteArrayString(d.values, value)
 	d.offsets = append(d.offsets, uint32(offset))
 	d.numValues++
-	return d.values[offset+plain.ByteArrayLengthSize : len(d.values) : len(d.values)]
+	return bits.BytesToString(d.values[offset+plain.ByteArrayLengthSize : len(d.values)])
 }
 
 func (d *byteArrayDictionary) Lookup(indexes []int32, values []Value) {
@@ -686,7 +738,19 @@ func (d *fixedLenByteArrayDictionary) value(i int32) []byte {
 }
 
 func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
-	_ = indexes[:len(values)]
+	d.insertValues(indexes, len(values), func(i int) *byte {
+		return values[i].ptr
+	})
+}
+
+func (d *fixedLenByteArrayDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	d.insertValues(indexes, rows.len, func(i int) *byte {
+		return (*byte)(rows.index(i, size, offset))
+	})
+}
+
+func (d *fixedLenByteArrayDictionary) insertValues(indexes []int32, count int, valueAt func(int) *byte) {
+	_ = indexes[:count]
 
 	if d.index == nil {
 		d.index = make(map[string]int32, cap(d.data)/d.size)
@@ -696,8 +760,8 @@ func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.ByteArray()
+	for i := 0; i < count; i++ {
+		value := unsafe.Slice(valueAt(i), d.size)
 
 		index, exists := d.index[string(value)]
 		if !exists {
@@ -769,7 +833,12 @@ func (d *uint32Dictionary) Len() int { return len(d.values) }
 func (d *uint32Dictionary) Index(i int32) Value { return makeValueUint32(d.values[i]) }
 
 func (d *uint32Dictionary) Insert(indexes []int32, values []Value) {
-	_ = indexes[:len(values)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *uint32Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[uint32]int32, cap(d.values))
@@ -778,8 +847,8 @@ func (d *uint32Dictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Uint32()
+	for i := 0; i < rows.len; i++ {
+		value := *(*uint32)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -850,7 +919,12 @@ func (d *uint64Dictionary) Len() int { return len(d.values) }
 func (d *uint64Dictionary) Index(i int32) Value { return makeValueUint64(d.values[i]) }
 
 func (d *uint64Dictionary) Insert(indexes []int32, values []Value) {
-	_ = indexes[:len(values)]
+	var value Value
+	d.insert(indexes, makeValueArray(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+}
+
+func (d *uint64Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
+	_ = indexes[:rows.len]
 
 	if d.index == nil {
 		d.index = make(map[uint64]int32, cap(d.values))
@@ -859,8 +933,8 @@ func (d *uint64Dictionary) Insert(indexes []int32, values []Value) {
 		}
 	}
 
-	for i, v := range values {
-		value := v.Uint64()
+	for i := 0; i < rows.len; i++ {
+		value := *(*uint64)(rows.index(i, size, offset))
 
 		index, exists := d.index[value]
 		if !exists {
@@ -1112,15 +1186,30 @@ func (col *indexedColumnBuffer) WriteValues(values []Value) (int, error) {
 	if j <= cap(col.values) {
 		col.values = col.values[:j]
 	} else {
-		colValues := make([]int32, j, 2*j)
-		copy(colValues, col.values)
-		col.values = colValues
+		tmp := make([]int32, j, 2*j)
+		copy(tmp, col.values)
+		col.values = tmp
 	}
 
 	col.typ.dict.Insert(col.values[i:], values)
 	return len(values), nil
 }
 
+func (col *indexedColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
+	i := len(col.values)
+	j := len(col.values) + rows.len
+
+	if j <= cap(col.values) {
+		col.values = col.values[:j]
+	} else {
+		tmp := make([]int32, j, 2*j)
+		copy(tmp, col.values)
+		col.values = tmp
+	}
+
+	col.typ.dict.insert(col.values[i:], rows, size, offset)
+}
+
 func (col *indexedColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
 	i := int(offset)
 	switch {
diff --git a/level.go b/level.go
new file mode 100644
index 0000000..911a946
--- /dev/null
+++ b/level.go
@@ -0,0 +1,27 @@
+package parquet
+
+import "github.com/segmentio/parquet-go/internal/bits"
+
+func countLevelsEqual(levels []byte, value byte) int {
+	return bits.CountByte(levels, value)
+}
+
+func countLevelsNotEqual(levels []byte, value byte) int {
+	return len(levels) - countLevelsEqual(levels, value)
+}
+
+func appendLevel(levels []byte, value byte, count int) []byte {
+	i := len(levels)
+	n := len(levels) + count
+
+	if cap(levels) < n {
+		newLevels := make([]byte, n, 2*n)
+		copy(newLevels, levels)
+		levels = newLevels
+	} else {
+		levels = levels[:n]
+	}
+
+	memset(levels[i:], value)
+	return levels
+}
diff --git a/level_amd64.go b/level_amd64.go
new file mode 100644
index 0000000..1ceec97
--- /dev/null
+++ b/level_amd64.go
@@ -0,0 +1,6 @@
+//go:build !purego
+
+package parquet
+
+//go:noescape
+func memset(dst []byte, src byte)
diff --git a/level_amd64.s b/level_amd64.s
new file mode 100644
index 0000000..a1d62a4
--- /dev/null
+++ b/level_amd64.s
@@ -0,0 +1,54 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// func memset(dst []byte, src byte)
+TEXT ·memset(SB), NOSPLIT, $0-32
+    MOVQ dst+0(FP), AX
+    MOVQ dst+8(FP), BX
+    MOVBQZX src+24(FP), CX
+
+    CMPQ BX, $8
+    JBE test
+
+    CMPQ BX, $64
+    JB init8
+
+    CMPB ·hasAVX2(SB), $0
+    JE init8
+
+    XORQ SI, SI
+    MOVQ BX, DX
+    SHRQ $6, DX
+    SHLQ $6, DX
+    MOVQ CX, X0
+    VPBROADCASTB X0, Y0
+loop64:
+    VMOVDQU Y0, (AX)(SI*1)
+    VMOVDQU Y0, 32(AX)(SI*1)
+    ADDQ $64, SI
+    CMPQ SI, DX
+    JNE loop64
+    VMOVDQU Y0, -64(AX)(BX*1)
+    VMOVDQU Y0, -32(AX)(BX*1)
+    VZEROUPPER
+    RET
+
+init8:
+    MOVQ $0x0101010101010101, R8
+    IMULQ R8, CX
+loop8:
+    MOVQ CX, -8(AX)(BX*1)
+    SUBQ $8, BX
+    CMPQ BX, $8
+    JAE loop8
+    MOVQ CX, (AX)
+    RET
+
+loop:
+    MOVB CX, -1(AX)(BX*1)
+    DECQ BX
+test:
+    CMPQ BX, $0
+    JNE loop
+    RET
diff --git a/level_purego.go b/level_purego.go
new file mode 100644
index 0000000..d72ff6c
--- /dev/null
+++ b/level_purego.go
@@ -0,0 +1,9 @@
+//go:build purego || !amd64
+
+package parquet
+
+func memset(dst []byte, src byte) {
+	for i := range dst {
+		dst[i] = src
+	}
+}
diff --git a/null.go b/null.go
new file mode 100644
index 0000000..2dc0019
--- /dev/null
+++ b/null.go
@@ -0,0 +1,111 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"bytes"
+	"reflect"
+	"unsafe"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/internal/unsafecast"
+)
+
+// nullIndexFunc is the type of functions used to detect null values in rows.
+//
+// For each value of the rows array, the bitmap passed as first argument is
+// populated to indicate whether the values were null (0) or not (1).
+//
+// The function writes one bit to the output buffer for each row in the input,
+// the buffer must be sized accordingly.
+type nullIndexFunc func(bits []uint64, rows array, size, offset uintptr)
+
+func nullIndex[T comparable](bits []uint64, rows array, size, offset uintptr) {
+	var zero T
+	for i := 0; i < rows.len; i++ {
+		v := *(*T)(rows.index(i, size, offset))
+		if v != zero {
+			x := uint(i) / 64
+			y := uint(i) % 64
+			bits[x] |= 1 << y
+		}
+	}
+}
+
+func nullIndexStruct(bits []uint64, rows array, size, offset uintptr) {
+	memset(unsafecast.Slice[byte](bits), 0xFF)
+}
+
+func nullIndexFuncOf(t reflect.Type) nullIndexFunc {
+	switch t {
+	case reflect.TypeOf(deprecated.Int96{}):
+		return nullIndex[deprecated.Int96]
+	}
+
+	switch t.Kind() {
+	case reflect.Bool:
+		return nullIndexBool
+
+	case reflect.Int:
+		return nullIndexInt
+
+	case reflect.Int32:
+		return nullIndexInt32
+
+	case reflect.Int64:
+		return nullIndexInt64
+
+	case reflect.Uint:
+		return nullIndexUint
+
+	case reflect.Uint32:
+		return nullIndexUint32
+
+	case reflect.Uint64:
+		return nullIndexUint64
+
+	case reflect.Float32:
+		return nullIndexFloat32
+
+	case reflect.Float64:
+		return nullIndexFloat64
+
+	case reflect.String:
+		return nullIndexString
+
+	case reflect.Slice:
+		return nullIndexSlice
+
+	case reflect.Array:
+		if t.Elem().Kind() == reflect.Uint8 {
+			switch size := t.Len(); size {
+			case 16:
+				return nullIndexUint128
+			default:
+				return nullIndexFuncOfByteArray(size)
+			}
+		}
+
+	case reflect.Pointer:
+		return nullIndexPointer
+
+	case reflect.Struct:
+		return nullIndexStruct
+	}
+
+	panic("cannot convert Go values of type " + t.String() + " to parquet value")
+}
+
+func nullIndexFuncOfByteArray(n int) nullIndexFunc {
+	return func(bits []uint64, rows array, size, offset uintptr) {
+		for i := 0; i < rows.len; i++ {
+			p := (*byte)(rows.index(i, size, offset))
+			b := unsafe.Slice(p, n)
+			if bytes.Count(b, []byte{0}) != n {
+				x := uint(i) / 64
+				y := uint(i) % 64
+				bits[x] |= 1 << y
+			}
+		}
+	}
+}
diff --git a/null_amd64.go b/null_amd64.go
new file mode 100644
index 0000000..19ad35f
--- /dev/null
+++ b/null_amd64.go
@@ -0,0 +1,72 @@
+//go:build go1.18 && !purego
+
+package parquet
+
+//go:noescape
+func nullIndex8bits(bits *uint64, rows array, size, offset uintptr)
+
+//go:noescape
+func nullIndex32bits(bits *uint64, rows array, size, offset uintptr)
+
+//go:noescape
+func nullIndex64bits(bits *uint64, rows array, size, offset uintptr)
+
+//go:noescape
+func nullIndex128bits(bits *uint64, rows array, size, offset uintptr)
+
+func nullIndexBool(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex8bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexInt(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexInt32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex32bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexInt64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexUint(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexUint32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex32bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexUint64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexUint128(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex128bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexFloat32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex32bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexFloat64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexString(bits []uint64, rows array, size, offset uintptr) {
+	// We offset by an extra 8 bytes to test the lengths of string values where
+	// the first field is the pointer and the second is the length which we want
+	// to test.
+	nullIndex64bits(&bits[0], rows, size, offset+8)
+}
+
+func nullIndexSlice(bits []uint64, rows array, size, offset uintptr) {
+	// Slice values are null if their pointer is nil, which is held in the first
+	// 8 bytes of the object so we can simply test 64 bits words.
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
+
+func nullIndexPointer(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex64bits(&bits[0], rows, size, offset)
+}
diff --git a/null_amd64.s b/null_amd64.s
new file mode 100644
index 0000000..6c45b6f
--- /dev/null
+++ b/null_amd64.s
@@ -0,0 +1,225 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// func nullIndex8bits(bits *uint64, rows array, size, offset uintptr)
+TEXT ·nullIndex8bits(SB), NOSPLIT, $0-40
+    MOVQ bits+0(FP), AX
+    MOVQ rows+8(FP), BX
+    MOVQ rows+16(FP), DI
+    MOVQ size+24(FP), DX
+    ADDQ offset+32(FP), BX
+
+    MOVQ $1, CX
+    XORQ SI, SI
+
+    CMPQ DI, $0
+    JE done
+loop1x1:
+    XORQ R8, R8
+    MOVB (BX), R9
+    CMPB R9, $0
+    JE next1x1
+
+    MOVQ SI, R10
+    SHRQ $6, R10
+    ORQ CX, (AX)(R10*8)
+next1x1:
+    ADDQ DX, BX
+    ROLQ $1, CX
+    INCQ SI
+    CMPQ SI, DI
+    JNE loop1x1
+done:
+    RET
+
+// func nullIndex32bits(bits *uint64, rows array, size, offset uintptr)
+TEXT ·nullIndex32bits(SB), NOSPLIT, $0-40
+    MOVQ bits+0(FP), AX
+    MOVQ rows+8(FP), BX
+    MOVQ rows+16(FP), DI
+    MOVQ size+24(FP), DX
+    ADDQ offset+32(FP), BX
+
+    MOVQ $1, CX
+    XORQ SI, SI
+
+    CMPQ DI, $0
+    JE done
+
+    CMPQ DI, $8
+    JB loop1x4
+
+    CMPB ·hasAVX2(SB), $0
+    JE loop1x4
+
+    MOVQ DI, R8
+    SHRQ $3, R8
+    SHLQ $3, R8
+
+    VPBROADCASTD size+24(FP), Y0
+    VPMULLD ·scale8x4(SB), Y0, Y0
+    VPCMPEQD Y1, Y1, Y1
+    VPCMPEQD Y2, Y2, Y2
+    VPXOR Y3, Y3, Y3
+loop8x4:
+    VPGATHERDD Y1, (BX)(Y0*1), Y4
+    VPCMPEQD Y3, Y4, Y4
+    VMOVMSKPS Y4, R9
+    VMOVDQU Y2, Y1
+
+    NOTQ R9
+    ANDQ $0b11111111, R9
+
+    MOVQ SI, CX
+    ANDQ $0b111111, CX
+
+    MOVQ SI, R10
+    SHRQ $6, R10
+
+    SHLQ CX, R9
+    ORQ R9, (AX)(R10*8)
+
+    LEAQ (BX)(DX*8), BX
+    ADDQ $8, SI
+    CMPQ SI, R8
+    JNE loop8x4
+    VZEROUPPER
+
+    CMPQ SI, DI
+    JE done
+
+    MOVQ $1, R8
+    MOVQ SI, CX
+    ANDQ $0b111111, R8
+    SHLQ CX, R8
+    MOVQ R8, CX
+
+loop1x4:
+    MOVL (BX), R8
+    CMPL R8, $0
+    JE next1x4
+
+    MOVQ SI, R9
+    SHRQ $6, R9
+    ORQ CX, (AX)(R9*8)
+next1x4:
+    ADDQ DX, BX
+    ROLQ $1, CX
+    INCQ SI
+    CMPQ SI, DI
+    JNE loop1x4
+done:
+    RET
+
+// func nullIndex64bits(bits *uint64, rows array, size, offset uintptr)
+TEXT ·nullIndex64bits(SB), NOSPLIT, $0-40
+    MOVQ bits+0(FP), AX
+    MOVQ rows+8(FP), BX
+    MOVQ rows+16(FP), DI
+    MOVQ size+24(FP), DX
+    ADDQ offset+32(FP), BX
+
+    MOVQ $1, CX
+    XORQ SI, SI
+
+    CMPQ DI, $0
+    JE done
+
+    CMPQ DI, $4
+    JB loop1x8
+
+    CMPB ·hasAVX2(SB), $0
+    JE loop1x8
+
+    MOVQ DI, R8
+    SHRQ $2, R8
+    SHLQ $2, R8
+
+    VPBROADCASTQ size+24(FP), Y0
+    VPMULLD ·scale4x8(SB), Y0, Y0
+    VPCMPEQQ Y1, Y1, Y1
+    VPCMPEQQ Y2, Y2, Y2
+    VPXOR Y3, Y3, Y3
+loop4x8:
+    VPGATHERQQ Y1, (BX)(Y0*1), Y4
+    VPCMPEQQ Y3, Y4, Y4
+    VMOVMSKPD Y4, R9
+    VMOVDQU Y2, Y1
+
+    NOTQ R9
+    ANDQ $0b1111, R9
+
+    MOVQ SI, CX
+    ANDQ $0b111111, CX
+
+    MOVQ SI, R10
+    SHRQ $6, R10
+
+    SHLQ CX, R9
+    ORQ R9, (AX)(R10*8)
+
+    LEAQ (BX)(DX*4), BX
+    ADDQ $4, SI
+    CMPQ SI, R8
+    JNE loop4x8
+    VZEROUPPER
+
+    CMPQ SI, DI
+    JE done
+
+    MOVQ $1, R8
+    MOVQ SI, CX
+    ANDQ $0b111111, R8
+    SHLQ CX, R8
+    MOVQ R8, CX
+
+loop1x8:
+    MOVQ (BX), R8
+    CMPQ R8, $0
+    JE next1x8
+
+    MOVQ SI, R9
+    SHRQ $6, R9
+    ORQ CX, (AX)(R9*8)
+next1x8:
+    ADDQ DX, BX
+    ROLQ $1, CX
+    INCQ SI
+    CMPQ SI, DI
+    JNE loop1x8
+done:
+    RET
+
+// func nullIndex128bits(bits *uint64, rows array, size, offset uintptr)
+TEXT ·nullIndex128bits(SB), NOSPLIT, $0-40
+    MOVQ bits+0(FP), AX
+    MOVQ rows+8(FP), BX
+    MOVQ rows+16(FP), DI
+    MOVQ size+24(FP), DX
+    ADDQ offset+32(FP), BX
+
+    CMPQ DI, $0
+    JE done
+
+    MOVQ $1, CX
+    XORQ SI, SI
+    PXOR X0, X0
+loop1x16:
+    MOVOU (BX), X1
+    PCMPEQQ X0, X1
+    MOVMSKPD X1, R8
+    CMPB R8, $0b11
+    JE next1x16
+
+    MOVQ SI, R9
+    SHRQ $6, R9
+    ORQ CX, (AX)(R9*8)
+next1x16:
+    ADDQ DX, BX
+    ROLQ $1, CX
+    INCQ SI
+    CMPQ SI, DI
+    JNE loop1x16
+done:
+    RET
diff --git a/null_purego.go b/null_purego.go
new file mode 100644
index 0000000..06a63b9
--- /dev/null
+++ b/null_purego.go
@@ -0,0 +1,62 @@
+//go:build go1.18 && (purego || !amd64)
+
+package parquet
+
+func nullIndexBool(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[bool](bits, rows, size, offset)
+}
+
+func nullIndexInt(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[int](bits, rows, size, offset)
+}
+
+func nullIndexInt32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[int32](bits, rows, size, offset)
+}
+
+func nullIndexInt64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[int64](bits, rows, size, offset)
+}
+
+func nullIndexUint(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[uint](bits, rows, size, offset)
+}
+
+func nullIndexUint32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[uint32](bits, rows, size, offset)
+}
+
+func nullIndexUint64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[uint64](bits, rows, size, offset)
+}
+
+func nullIndexUint128(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[[16]byte](bits, rows, size, offset)
+}
+
+func nullIndexFloat32(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[float32](bits, rows, size, offset)
+}
+
+func nullIndexFloat64(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[float64](bits, rows, size, offset)
+}
+
+func nullIndexString(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[string](bits, rows, size, offset)
+}
+
+func nullIndexSlice(bits []uint64, rows array, size, offset uintptr) {
+	for i := 0; i < rows.len; i++ {
+		p := *(**struct{})(rows.index(i, size, offset))
+		b := uint64(0)
+		if p != nil {
+			b = 1
+		}
+		bits[uint(i)/64] |= b << (uint(i) % 64)
+	}
+}
+
+func nullIndexPointer(bits []uint64, rows array, size, offset uintptr) {
+	nullIndex[*struct{}](bits, rows, size, offset)
+}
diff --git a/page.go b/page.go
index 3b4d4f4..a3aae73 100644
--- a/page.go
+++ b/page.go
@@ -275,35 +275,6 @@ func errPageBoundsOutOfRange(i, j, n int64) error {
 	return fmt.Errorf("page bounds out of range [%d:%d]: with length %d", i, j, n)
 }
 
-func countLevelsEqual(levels []byte, value byte) int {
-	return bits.CountByte(levels, value)
-}
-
-func countLevelsNotEqual(levels []byte, value byte) int {
-	return len(levels) - countLevelsEqual(levels, value)
-}
-
-func appendLevel(levels []byte, value byte, count int) []byte {
-	if count > 0 {
-		i := len(levels)
-		n := len(levels) + count
-
-		if cap(levels) < n {
-			newLevels := make([]byte, n)
-			copy(newLevels, levels)
-			levels = newLevels
-		} else {
-			levels = levels[:n]
-		}
-
-		fill := levels[i:]
-		for i := range fill {
-			fill[i] = value
-		}
-	}
-	return levels
-}
-
 type optionalPage struct {
 	base               BufferedPage
 	maxDefinitionLevel byte
diff --git a/parquet_amd64.go b/parquet_amd64.go
new file mode 100644
index 0000000..e36a2c1
--- /dev/null
+++ b/parquet_amd64.go
@@ -0,0 +1,11 @@
+//go:build !purego
+
+package parquet
+
+import "golang.org/x/sys/cpu"
+
+var (
+	// This variable is used in x86 assembly source files to gate the use of
+	// AVX2 instructions depending on whether the CPU supports it.
+	hasAVX2 = cpu.X86.HasAVX2
+)
diff --git a/writer_go18.go b/writer_go18.go
index fe81a75..b596c70 100644
--- a/writer_go18.go
+++ b/writer_go18.go
@@ -51,7 +51,7 @@ type GenericWriter[T any] struct {
 	// schema of the parquet file.
 	write writeFunc[T]
 	// This field is used to leverage the optimized writeRowsFunc algorithms.
-	buffers columnBufferWriter
+	columns []ColumnBuffer
 }
 
 // NewGenericWriter is like NewWriter but returns a GenericWriter[T] suited to
@@ -109,8 +109,17 @@ func makeWriteFunc[T any](t reflect.Type, schema *Schema) writeFunc[T] {
 	size := t.Size()
 	writeRows := writeRowsFuncOf(t, schema, nil)
 	return func(w *GenericWriter[T], rows []T) (n int, err error) {
-		defer w.buffers.clear()
-		err = writeRows(&w.buffers, makeArray(rows), size, 0, columnLevels{})
+		if w.columns == nil {
+			w.columns = make([]ColumnBuffer, len(w.base.writer.columns))
+			for i, c := range w.base.writer.columns {
+				// These fields are usually lazily initialized when writing rows,
+				// we need them to exist now tho.
+				c.columnBuffer = c.newColumnBuffer()
+				c.maxValues = int32(c.columnBuffer.Cap())
+				w.columns[i] = c.columnBuffer
+			}
+		}
+		err = writeRows(w.columns, makeArray(rows), size, 0, columnLevels{})
 		if err == nil {
 			n = len(rows)
 		}
@@ -131,21 +140,6 @@ func (w *GenericWriter[T]) Reset(output io.Writer) {
 }
 
 func (w *GenericWriter[T]) Write(rows []T) (int, error) {
-	if w.buffers.columns == nil {
-		w.buffers = columnBufferWriter{
-			columns: make([]ColumnBuffer, len(w.base.writer.columns)),
-			values:  make([]Value, 0, defaultValueBufferSize),
-		}
-
-		for i, c := range w.base.writer.columns {
-			// These fields are usually lazily initialized when writing rows,
-			// we need them to exist now tho.
-			c.columnBuffer = c.newColumnBuffer()
-			c.maxValues = int32(c.columnBuffer.Cap())
-			w.buffers.columns[i] = c.columnBuffer
-		}
-	}
-
 	n, err := w.write(w, rows)
 	if err != nil {
 		return n, err


Test patch:
diff --git a/buffer_go18_test.go b/buffer_go18_test.go
index 7848a52..f5eef40 100644
--- a/buffer_go18_test.go
+++ b/buffer_go18_test.go
@@ -36,6 +36,7 @@ func TestGenericBuffer(t *testing.T) {
 	testGenericBuffer[nestedListColumn](t)
 	testGenericBuffer[*contact](t)
 	testGenericBuffer[paddedBooleanColumn](t)
+	testGenericBuffer[optionalInt32Column](t)
 }
 
 func testGenericBuffer[Row any](t *testing.T) {
@@ -110,6 +111,7 @@ func BenchmarkGenericBuffer(b *testing.B) {
 	benchmarkGenericBuffer[decimalColumn](b)
 	benchmarkGenericBuffer[contact](b)
 	benchmarkGenericBuffer[paddedBooleanColumn](b)
+	benchmarkGenericBuffer[optionalInt32Column](b)
 }
 
 func benchmarkGenericBuffer[Row generator[Row]](b *testing.B) {
diff --git a/buffer_test.go b/buffer_test.go
index e1c4d62..017c043 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -551,6 +551,40 @@ func TestBufferSeekToRow(t *testing.T) {
 	}
 }
 
+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.WriteRows([]parquet.Row{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)
+	}
+}
+
 func generateBenchmarkBufferRows(n int) (*parquet.Schema, []parquet.Row) {
 	model := new(benchmarkRowType)
 	schema := parquet.SchemaOf(model)
diff --git a/column_buffer_test.go b/column_buffer_test.go
deleted file mode 100644
index 1ac22ae..0000000
--- a/column_buffer_test.go
+++ /dev/null
@@ -1,42 +0,0 @@
-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.WriteRows([]parquet.Row{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/level_test.go b/level_test.go
new file mode 100644
index 0000000..050748e
--- /dev/null
+++ b/level_test.go
@@ -0,0 +1,43 @@
+package parquet
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestMemset(t *testing.T) {
+	const N = 100_0000
+	buffer := make([]byte, N)
+
+	for n := 1; n <= N; n = (n * 2) + 1 {
+		t.Run(fmt.Sprintf("size=%d", n), func(t *testing.T) {
+			b := buffer[:n]
+
+			for i := range b {
+				b[i] = 0
+			}
+
+			memset(b, 42)
+
+			for i, c := range b {
+				if c != 42 {
+					t.Fatalf("byte at index %d has value %d", i, c)
+				}
+			}
+		})
+	}
+}
+
+func BenchmarkMemset(b *testing.B) {
+	for _, size := range []int{0, 10, 100, 1000, 10_000} {
+		b.Run(fmt.Sprintf("size=%d", size), func(b *testing.B) {
+			data := make([]byte, size)
+
+			for i := 0; i < b.N; i++ {
+				memset(data, 1)
+			}
+
+			b.SetBytes(int64(size))
+		})
+	}
+}
diff --git a/null_test.go b/null_test.go
new file mode 100644
index 0000000..462bb67
--- /dev/null
+++ b/null_test.go
@@ -0,0 +1,100 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"reflect"
+	"testing"
+	"unsafe"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/internal/quick"
+)
+
+func TestNullIndex(t *testing.T) {
+	testNullIndex[bool](t)
+	testNullIndex[int](t)
+	testNullIndex[int32](t)
+	testNullIndex[int64](t)
+	testNullIndex[uint](t)
+	testNullIndex[uint32](t)
+	testNullIndex[uint64](t)
+	testNullIndex[float32](t)
+	testNullIndex[float64](t)
+	testNullIndex[[10]byte](t)
+	testNullIndex[[16]byte](t)
+	testNullIndex[deprecated.Int96](t)
+	testNullIndex[string](t)
+	testNullIndex[*struct{}](t)
+}
+
+func testNullIndex[T comparable](t *testing.T) {
+	var zero T
+	t.Helper()
+	t.Run(reflect.TypeOf(zero).String(), func(t *testing.T) {
+		err := quick.Check(func(data []T) bool {
+			if len(data) == 0 {
+				return true
+			}
+
+			size := unsafe.Sizeof(zero)
+			want := make([]uint64, (len(data)+63)/64)
+			got := make([]uint64, (len(data)+63)/64)
+
+			for i := range data {
+				if (i % 2) == 0 {
+					data[i] = zero
+				}
+			}
+
+			array := makeArray(data)
+			nullIndex[T](want, array, size, 0)
+			nullIndexFuncOf(reflect.TypeOf(zero))(got, array, size, 0)
+
+			if !reflect.DeepEqual(want, got) {
+				t.Errorf("unexpected null index\nwant = %064b\ngot  = %064b", want, got)
+				return false
+			}
+			return true
+		})
+		if err != nil {
+			t.Error(err)
+		}
+	})
+}
+
+func BenchmarkNullIndex(b *testing.B) {
+	benchmarkNullIndex[bool](b)
+	benchmarkNullIndex[int](b)
+	benchmarkNullIndex[int32](b)
+	benchmarkNullIndex[int64](b)
+	benchmarkNullIndex[uint](b)
+	benchmarkNullIndex[uint32](b)
+	benchmarkNullIndex[uint64](b)
+	benchmarkNullIndex[float32](b)
+	benchmarkNullIndex[float64](b)
+	benchmarkNullIndex[[10]byte](b)
+	benchmarkNullIndex[[16]byte](b)
+	benchmarkNullIndex[deprecated.Int96](b)
+	benchmarkNullIndex[string](b)
+	benchmarkNullIndex[[]struct{}](b)
+	benchmarkNullIndex[*struct{}](b)
+}
+
+func benchmarkNullIndex[T any](b *testing.B) {
+	const N = 1000
+
+	var zero T
+	typ := reflect.TypeOf(zero)
+	size := unsafe.Sizeof(zero)
+	null := nullIndexFuncOf(typ)
+	data := makeArray(make([]T, N))
+	bits := make([]uint64, (N+63)/64)
+
+	b.Run(typ.String(), func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			null(bits, data, size, 0)
+		}
+		b.SetBytes(int64(size * N))
+	})
+}
diff --git a/parquet_test.go b/parquet_test.go
index ba20232..4b1a7aa 100644
--- a/parquet_test.go
+++ b/parquet_test.go
@@ -175,6 +175,14 @@ func (row contact) generate(prng *rand.Rand) contact {
 	}
 }
 
+type optionalInt32Column struct {
+	Value int32 `parquet:",optional"`
+}
+
+func (row optionalInt32Column) generate(prng *rand.Rand) optionalInt32Column {
+	return optionalInt32Column{Value: prng.Int31n(100)}
+}
+
 type listColumn2 struct {
 	Value utf8string `parquet:",optional"`
 }
diff --git a/reader_go18_test.go b/reader_go18_test.go
index 0bbb66c..cf5f07a 100644
--- a/reader_go18_test.go
+++ b/reader_go18_test.go
@@ -36,6 +36,8 @@ func TestGenericReader(t *testing.T) {
 	testGenericReader[nestedListColumn1](t)
 	testGenericReader[nestedListColumn](t)
 	testGenericReader[*contact](t)
+	testGenericReader[paddedBooleanColumn](t)
+	testGenericReader[optionalInt32Column](t)
 }
 
 func testGenericReader[Row any](t *testing.T) {
@@ -98,6 +100,8 @@ func BenchmarkGenericReader(b *testing.B) {
 	benchmarkGenericReader[mapColumn](b)
 	benchmarkGenericReader[decimalColumn](b)
 	benchmarkGenericReader[contact](b)
+	benchmarkGenericReader[paddedBooleanColumn](b)
+	benchmarkGenericReader[optionalInt32Column](b)
 }
 
 func benchmarkGenericReader[Row generator[Row]](b *testing.B) {
diff --git a/writer_go18_test.go b/writer_go18_test.go
index 8893162..d7b9208 100644
--- a/writer_go18_test.go
+++ b/writer_go18_test.go
@@ -26,6 +26,8 @@ func BenchmarkGenericWriter(b *testing.B) {
 	benchmarkGenericWriter[mapColumn](b)
 	benchmarkGenericWriter[decimalColumn](b)
 	benchmarkGenericWriter[contact](b)
+	benchmarkGenericWriter[paddedBooleanColumn](b)
+	benchmarkGenericWriter[optionalInt32Column](b)
 }
 
 func benchmarkGenericWriter[Row generator[Row]](b *testing.B) {