Repository: segmentio/parquet-go
Pull request #260 (2022-07-06 23:03:38)

Optimize dictionaries for fixed-length types
This PR contributes to https://github.com/segmentio/parquet-go/issues/226 by adding a new `hashprobe` package which provides optimized implementations of probing tables, and using these data structures as replacement for the use of Go maps to back parquet dictionaries.

The gains come from two key observations:
* the Go map API does not offer the ability to perform bulk operations, resulting in spending a lot of CPU time calling into the runtime map functions
* the Go map implementation has to make performance trade offs to support operations like deletion and iteration, which we do not need in parquet dictionary, creating the opportunity to use a more efficient memory layout

I also added two subpackages: `hashprobe/aeshash` and `hashprobe/wyhash` providing implementations of hash functions similar to the ones used by the Go runtime.

The change measurable improvements compared to the use of Go maps:
```
name                                        old time/op  new time/op  delta
Dictionary/Insert/INT32                     7.37µs ± 0%  2.47µs ± 0%   -66.49%  (p=0.000 n=9+9)
Dictionary/Insert/INT64                     7.48µs ± 1%  2.69µs ± 0%   -64.00%  (p=0.000 n=10+10)
Dictionary/Insert/FLOAT                     21.1µs ± 1%   2.4µs ± 0%   -88.52%  (p=0.000 n=10+9)
Dictionary/Insert/DOUBLE                    22.7µs ± 0%   2.8µs ± 1%   -87.57%  (p=0.000 n=10+10)
Dictionary/Insert/FIXED_LEN_BYTE_ARRAY(16)  21.5µs ± 1%   4.9µs ± 0%   -77.19%  (p=0.000 n=10+10)
Dictionary/Insert/INT(32,false)             7.49µs ± 1%  2.52µs ± 0%   -66.39%  (p=0.000 n=10+10)
Dictionary/Insert/INT(64,false)             7.43µs ± 1%  2.76µs ± 1%   -62.83%  (p=0.000 n=10+10)

name                                        old value/s  new value/s  delta
Dictionary/Insert/INT32                       136M ± 0%    405M ± 0%  +198.33%  (p=0.000 n=9+10)
Dictionary/Insert/INT64                       134M ± 1%    371M ± 0%  +177.77%  (p=0.000 n=10+10)
Dictionary/Insert/FLOAT                      47.5M ± 1%  413.4M ± 0%  +770.91%  (p=0.000 n=10+9)
Dictionary/Insert/DOUBLE                     44.1M ± 0%  354.5M ± 1%  +704.21%  (p=0.000 n=10+10)
Dictionary/Insert/FIXED_LEN_BYTE_ARRAY(16)   46.4M ± 1%  203.5M ± 0%  +338.37%  (p=0.000 n=10+10)
Dictionary/Insert/INT(32,false)               133M ± 1%    397M ± 0%  +197.51%  (p=0.000 n=10+10)
Dictionary/Insert/INT(64,false)               135M ± 1%    363M ± 0%  +169.27%  (p=0.000 n=10+9)
```

I have a few follow up but I think this PR has enough content to be reviewed already:
* validation of `parquet.Value` inserted to BE128 dictionaries is important but costs over 40% of the compute footprint, this looks like a good are for improvement considering this type would be often used for dictionaries of UUID-like values
* I have not yet investigated dictionaries of variable-length values (e.g. `string`), but these are very popular models in parquet so worth providing a specialization for


Patch:
diff --git a/array.go b/array.go
index 2038e2b..86c0b6f 100644
--- a/array.go
+++ b/array.go
@@ -2,46 +2,22 @@ package parquet
 
 import (
 	"unsafe"
-)
-
-type array struct {
-	ptr unsafe.Pointer
-	len int
-}
 
-func makeArray(ptr unsafe.Pointer, len int) array {
-	return array{ptr: ptr, len: len}
-}
-
-func makeArrayBool(values []bool) array {
-	return *(*array)(unsafe.Pointer(&values))
-}
-
-func makeArrayString(values []string) array {
-	return *(*array)(unsafe.Pointer(&values))
-}
-
-func makeArrayValue(values []Value) array {
-	return *(*array)(unsafe.Pointer(&values))
-}
+	"github.com/segmentio/parquet-go/sparse"
+)
 
-func makeArrayBE128(values []*[16]byte) array {
-	return *(*array)(unsafe.Pointer(&values))
+func makeArrayValue(values []Value, offset uintptr) sparse.Array {
+	ptr := *(*unsafe.Pointer)(unsafe.Pointer(&values))
+	return sparse.UnsafeArray(unsafe.Add(ptr, offset), len(values), unsafe.Sizeof(Value{}))
 }
 
-func (a array) index(i int, size, offset uintptr) unsafe.Pointer {
-	return unsafe.Add(a.ptr, uintptr(i)*size+offset)
+func makeArrayString(values []string) sparse.Array {
+	str := ""
+	ptr := *(*unsafe.Pointer)(unsafe.Pointer(&values))
+	return sparse.UnsafeArray(ptr, len(values), unsafe.Sizeof(str))
 }
 
-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,
-	}
+func makeArrayBE128(values []*[16]byte) sparse.Array {
+	ptr := *(*unsafe.Pointer)(unsafe.Pointer(&values))
+	return sparse.UnsafeArray(ptr, len(values), unsafe.Sizeof((*[16]byte)(nil)))
 }
diff --git a/array_go18.go b/array_go18.go
index c912946..1ce1a98 100644
--- a/array_go18.go
+++ b/array_go18.go
@@ -6,16 +6,28 @@ import (
 	"unsafe"
 
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
-func makeArrayOf[T any](s []T) array {
-	return makeArray(unsafecast.PointerOf(s), len(s))
+func makeArray(base unsafe.Pointer, length int, offset uintptr) sparse.Array {
+	return sparse.UnsafeArray(base, length, offset)
 }
 
-func makeSlice[T any](a array) []T {
-	return slice[T](a.ptr, a.len)
+func makeArrayOf[T any](s []T) sparse.Array {
+	var model T
+	return makeArray(unsafecast.PointerOf(s), len(s), unsafe.Sizeof(model))
+}
+
+func makeSlice[T any](a sparse.Array) []T {
+	return slice[T](a.Index(0), a.Len())
 }
 
 func slice[T any](p unsafe.Pointer, n int) []T {
 	return unsafe.Slice((*T)(p), n)
 }
+
+type sliceHeader struct {
+	base unsafe.Pointer
+	len  int
+	cap  int
+}
diff --git a/buffer_go18.go b/buffer_go18.go
index 3a28a08..e393928 100644
--- a/buffer_go18.go
+++ b/buffer_go18.go
@@ -69,10 +69,9 @@ func bufferFuncOf[T any](t reflect.Type, schema *Schema) bufferFunc[T] {
 }
 
 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) {
-		err = writeRows(buf.base.columns, makeArrayOf(rows), size, 0, columnLevels{})
+		err = writeRows(buf.base.columns, makeArrayOf(rows), columnLevels{})
 		if err == nil {
 			n = len(rows)
 		}
diff --git a/column_buffer.go b/column_buffer.go
index f3587ac..9981831 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -11,6 +11,7 @@ import (
 	"github.com/segmentio/parquet-go/encoding/plain"
 	"github.com/segmentio/parquet-go/internal/bitpack"
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
 // ColumnBuffer is an interface representing columns of a row group.
@@ -77,7 +78,7 @@ type ColumnBuffer interface {
 	// 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)
+	writeValues(rows sparse.Array, levels columnLevels)
 }
 
 type columnLevels struct {
@@ -331,20 +332,20 @@ func (col *optionalColumnBuffer) WriteValues(values []Value) (n int, err error)
 	return n, nil
 }
 
-func (col *optionalColumnBuffer) writeValues(rows array, size, offset uintptr, levels columnLevels) {
+func (col *optionalColumnBuffer) writeValues(rows sparse.Array, 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 {
+	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)
+	col.definitionLevels = appendLevel(col.definitionLevels, levels.definitionLevel, rows.Len())
 
 	i := len(col.rows)
-	j := len(col.rows) + rows.len
+	j := len(col.rows) + rows.Len()
 
 	if j <= cap(col.rows) {
 		col.rows = col.rows[:j]
@@ -358,7 +359,7 @@ func (col *optionalColumnBuffer) writeValues(rows array, size, offset uintptr, l
 		broadcastValueInt32(col.rows[i:], -1)
 	} else {
 		broadcastRangeInt32(col.rows[i:], int32(col.base.Len()))
-		col.base.writeValues(rows, size, offset, levels)
+		col.base.writeValues(rows, levels)
 	}
 }
 
@@ -679,7 +680,7 @@ func (col *repeatedColumnBuffer) writeRow(row []Value) error {
 	return nil
 }
 
-func (col *repeatedColumnBuffer) writeValues(row array, size, offset uintptr, levels columnLevels) {
+func (col *repeatedColumnBuffer) writeValues(row sparse.Array, levels columnLevels) {
 	if levels.repetitionLevel == 0 {
 		col.rows = append(col.rows, region{
 			offset:     uint32(len(col.repetitionLevels)),
@@ -687,17 +688,17 @@ func (col *repeatedColumnBuffer) writeValues(row array, size, offset uintptr, le
 		})
 	}
 
-	if row.len == 0 {
+	if row.Len() == 0 {
 		col.repetitionLevels = append(col.repetitionLevels, levels.repetitionLevel)
 		col.definitionLevels = append(col.definitionLevels, levels.definitionLevel)
 		return
 	}
 
-	col.repetitionLevels = appendLevel(col.repetitionLevels, levels.repetitionLevel, row.len)
-	col.definitionLevels = appendLevel(col.definitionLevels, levels.definitionLevel, row.len)
+	col.repetitionLevels = appendLevel(col.repetitionLevels, levels.repetitionLevel, row.Len())
+	col.definitionLevels = appendLevel(col.definitionLevels, levels.definitionLevel, row.Len())
 
 	if levels.definitionLevel == col.maxDefinitionLevel {
-		col.base.writeValues(row, size, offset, levels)
+		col.base.writeValues(row, levels)
 	}
 }
 
@@ -799,33 +800,34 @@ func (col *booleanColumnBuffer) Swap(i, j int) {
 }
 
 func (col *booleanColumnBuffer) WriteBooleans(values []bool) (int, error) {
-	col.writeValues(makeArrayBool(values), unsafe.Sizeof(false), 0, columnLevels{})
+	col.writeValues(sparse.MakeBoolArray(values).UnsafeArray(), columnLevels{})
 	return len(values), nil
 }
 
 func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *booleanColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	numBytes := bitpack.ByteCount(uint(col.numValues) + uint(rows.len))
+func (col *booleanColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
+	numBytes := bitpack.ByteCount(uint(col.numValues) + uint(rows.Len()))
 	if cap(col.bits) < numBytes {
 		col.bits = append(make([]byte, 0, 2*cap(col.bits)), col.bits...)
 	}
 	col.bits = col.bits[:numBytes]
 	i := 0
 	r := 8 - (int(col.numValues) % 8)
+	bytes := rows.Uint8Array()
 
-	if r <= rows.len {
+	if r <= bytes.Len() {
 		// First we attempt to write enough bits to align the number of values
 		// in the column buffer on 8 bytes. After this step the next bit should
 		// be written at the zero'th index of a byte of the buffer.
 		if r < 8 {
 			var b byte
 			for i < r {
-				v := *(*byte)(rows.index(i, size, offset))
+				v := bytes.Index(i)
 				b |= (v & 1) << uint(i)
 				i++
 			}
@@ -835,50 +837,22 @@ func (col *booleanColumnBuffer) writeValues(rows array, size, offset uintptr, _
 			col.numValues += int32(i)
 		}
 
-		if n := ((rows.len - i) / 8) * 8; n > 0 {
+		if n := ((bytes.Len() - i) / 8) * 8; n > 0 {
 			// At this stage, we know that that we have at least 8 bits to write
 			// and the bits will be aligned on the address of a byte in the
 			// output buffer. We can work on 8 values per loop iteration,
 			// packing them into a single byte and writing it to the output
 			// buffer. This effectively reduces by 87.5% the number of memory
 			// stores that the program needs to perform to generate the values.
-			if optimize(n) {
-				section := array{
-					ptr: rows.index(i, size, 0),
-					len: n,
-				}
-				writeValuesBool(col.bits[col.numValues/8:], section, size, offset)
-				i += section.len
-			} else {
-				for j := col.numValues / 8; i < n; j++ {
-					b0 := *(*byte)(rows.index(i+0, size, offset))
-					b1 := *(*byte)(rows.index(i+1, size, offset))
-					b2 := *(*byte)(rows.index(i+2, size, offset))
-					b3 := *(*byte)(rows.index(i+3, size, offset))
-					b4 := *(*byte)(rows.index(i+4, size, offset))
-					b5 := *(*byte)(rows.index(i+5, size, offset))
-					b6 := *(*byte)(rows.index(i+6, size, offset))
-					b7 := *(*byte)(rows.index(i+7, size, offset))
-
-					col.bits[j] = (b0 & 1) |
-						((b1 & 1) << 1) |
-						((b2 & 1) << 2) |
-						((b3 & 1) << 3) |
-						((b4 & 1) << 4) |
-						((b5 & 1) << 5) |
-						((b6 & 1) << 6) |
-						((b7 & 1) << 7)
-					i += 8
-				}
-			}
+			i += sparse.GatherBits(col.bits[col.numValues/8:], bytes.Slice(i, i+n))
 			col.numValues += int32(n)
 		}
 	}
 
-	for i < rows.len {
+	for i < bytes.Len() {
 		x := uint(col.numValues) / 8
 		y := uint(col.numValues) % 8
-		b := *(*byte)(rows.index(i, size, offset))
+		b := bytes.Index(i)
 		col.bits[x] = ((b & 1) << y) | (col.bits[x] & ^(1 << y))
 		col.numValues++
 		i++
@@ -967,26 +941,19 @@ func (col *int32ColumnBuffer) WriteInt32s(values []int32) (int, error) {
 }
 
 func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *int32ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *int32ColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
-
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherInt32(col.values[n:], rows.Int32Array())
 
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesInt32(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*int32)(rows.index(i, size, offset))
-		}
-	}
 }
 
 func (col *int32ColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1069,25 +1036,18 @@ func (col *int64ColumnBuffer) WriteInt64s(values []int64) (int, error) {
 }
 
 func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *int64ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *int64ColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesInt64(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*int64)(rows.index(i, size, offset))
-		}
-	}
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherInt64(col.values[n:], rows.Int64Array())
 }
 
 func (col *int64ColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1176,9 +1136,9 @@ func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
 	return len(values), nil
 }
 
-func (col *int96ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	for i := 0; i < rows.len; i++ {
-		p := rows.index(i, size, offset)
+func (col *int96ColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
+	for i := 0; i < rows.Len(); i++ {
+		p := rows.Index(i)
 		col.values = append(col.values, *(*deprecated.Int96)(p))
 	}
 }
@@ -1263,25 +1223,18 @@ func (col *floatColumnBuffer) WriteFloats(values []float32) (int, error) {
 }
 
 func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *floatColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *floatColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesFloat32(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*float32)(rows.index(i, size, offset))
-		}
-	}
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherFloat32(col.values[n:], rows.Float32Array())
 }
 
 func (col *floatColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1364,25 +1317,18 @@ func (col *doubleColumnBuffer) WriteDoubles(values []float64) (int, error) {
 }
 
 func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *doubleColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *doubleColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesFloat64(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*float64)(rows.index(i, size, offset))
-		}
-	}
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherFloat64(col.values[n:], rows.Float64Array())
 }
 
 func (col *doubleColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1516,14 +1462,14 @@ func (col *byteArrayColumnBuffer) writeByteArrays(values []byte) (count, bytes i
 }
 
 func (col *byteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.ptr), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.ptr)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *byteArrayColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	for i := 0; i < rows.len; i++ {
-		p := rows.index(i, size, offset)
+func (col *byteArrayColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
+	for i := 0; i < rows.Len(); i++ {
+		p := rows.Index(i)
 		col.append(*(*string)(p))
 	}
 }
@@ -1644,8 +1590,8 @@ func (col *fixedLenByteArrayColumnBuffer) WriteValues(values []Value) (int, erro
 	return len(values), nil
 }
 
-func (col *fixedLenByteArrayColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	n := col.size * rows.len
+func (col *fixedLenByteArrayColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
+	n := col.size * rows.Len()
 	i := len(col.data)
 	j := len(col.data) + n
 
@@ -1656,8 +1602,8 @@ func (col *fixedLenByteArrayColumnBuffer) writeValues(rows array, size, offset u
 	col.data = col.data[:j]
 	newData := col.data[i:]
 
-	for i := 0; i < rows.len; i++ {
-		p := rows.index(i, size, offset)
+	for i := 0; i < rows.Len(); i++ {
+		p := rows.Index(i)
 		copy(newData[i*col.size:], unsafe.Slice((*byte)(p), col.size))
 	}
 }
@@ -1742,25 +1688,18 @@ func (col *uint32ColumnBuffer) WriteUint32s(values []uint32) (int, error) {
 }
 
 func (col *uint32ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *uint32ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *uint32ColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesUint32(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*uint32)(rows.index(i, size, offset))
-		}
-	}
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherUint32(col.values[n:], rows.Uint32Array())
 }
 
 func (col *uint32ColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1843,25 +1782,18 @@ func (col *uint64ColumnBuffer) WriteUint64s(values []uint64) (int, error) {
 }
 
 func (col *uint64ColumnBuffer) WriteValues(values []Value) (int, error) {
-	var value Value
-	col.writeValues(makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64), columnLevels{})
+	var model Value
+	col.writeValues(makeArrayValue(values, unsafe.Offsetof(model.u64)), columnLevels{})
 	return len(values), nil
 }
 
-func (col *uint64ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *uint64ColumnBuffer) writeValues(rows sparse.Array, _ 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...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-
-	if values := col.values[n:]; optimize(len(values)) {
-		writeValuesUint64(values, rows, size, offset)
-	} else {
-		for i := range values {
-			values[i] = *(*uint64)(rows.index(i, size, offset))
-		}
-	}
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherUint64(col.values[n:], rows.Uint64Array())
 }
 
 func (col *uint64ColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
@@ -1949,13 +1881,13 @@ func (col *be128ColumnBuffer) WriteValues(values []Value) (int, error) {
 	return len(values), nil
 }
 
-func (col *be128ColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
-	if n := len(col.values) + rows.len; n > cap(col.values) {
+func (col *be128ColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
+	if n := len(col.values) + rows.Len(); n > cap(col.values) {
 		col.values = append(make([][16]byte, 0, max(n, 2*cap(col.values))), col.values...)
 	}
 	n := len(col.values)
-	col.values = col.values[:n+rows.len]
-	writeValuesBE128(col.values[n:], rows, size, offset)
+	col.values = col.values[:n+rows.Len()]
+	sparse.GatherUint128(col.values[n:], rows.Uint128Array())
 }
 
 func (col *be128ColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
diff --git a/column_buffer_amd64.go b/column_buffer_amd64.go
index 8c34fa8..05a0722 100644
--- a/column_buffer_amd64.go
+++ b/column_buffer_amd64.go
@@ -3,10 +3,9 @@
 package parquet
 
 import (
-	"unsafe"
-
 	"github.com/segmentio/parquet-go/internal/bytealg"
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 	"golang.org/x/sys/cpu"
 )
 
@@ -18,7 +17,7 @@ func broadcastValueInt32(dst []int32, src int8) {
 func broadcastRangeInt32AVX2(dst []int32, base int32)
 
 func broadcastRangeInt32(dst []int32, base int32) {
-	if len(dst) >= minLenAVX2 && cpu.X86.HasAVX2 {
+	if len(dst) >= 8 && cpu.X86.HasAVX2 {
 		broadcastRangeInt32AVX2(dst, base)
 	} else {
 		for i := range dst {
@@ -28,45 +27,4 @@ func broadcastRangeInt32(dst []int32, base int32) {
 }
 
 //go:noescape
-func writeValuesBitpackAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-
-//go:noescape
-func writeValues32bitsAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-
-//go:noescpae
-func writeValues64bitsAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-
-//go:noescape
-func writeValues128bits(values unsafe.Pointer, rows array, size, offset uintptr)
-
-func writeValuesBool(values []byte, rows array, size, offset uintptr) {
-	writeValuesBitpackAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesInt32(values []int32, rows array, size, offset uintptr) {
-	writeValues32bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesInt64(values []int64, rows array, size, offset uintptr) {
-	writeValues64bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesUint32(values []uint32, rows array, size, offset uintptr) {
-	writeValues32bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesUint64(values []uint64, rows array, size, offset uintptr) {
-	writeValues64bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesFloat32(values []float32, rows array, size, offset uintptr) {
-	writeValues32bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesFloat64(values []float64, rows array, size, offset uintptr) {
-	writeValues64bitsAVX2(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
-
-func writeValuesBE128(values [][16]byte, rows array, size, offset uintptr) {
-	writeValues128bits(*(*unsafe.Pointer)(unsafe.Pointer(&values)), rows, size, offset)
-}
+func writePointersBE128(values [][16]byte, rows sparse.Array)
diff --git a/column_buffer_amd64.s b/column_buffer_amd64.s
index d83d50d..1eef03d 100644
--- a/column_buffer_amd64.s
+++ b/column_buffer_amd64.s
@@ -41,237 +41,27 @@ test1x4:
     JNE loop1x4
     RET
 
-// func writeValuesBitpackAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-TEXT ·writeValuesBitpackAVX2(SB), NOSPLIT, $0-40
-    MOVQ values+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), CX
-    MOVQ size+24(FP), DX
-    MOVQ offset+32(FP), DI
+// func writePointersBE128(values [][16]byte, rows sparse.Array)
+TEXT ·writePointersBE128(SB), NOSPLIT, $0-48
+    MOVQ values_base+0(FP), AX
+    MOVQ rows_array_ptr+24(FP), BX
+    MOVQ rows_array_len+32(FP), CX
+    MOVQ rows_array_off+40(FP), DX
 
-    CMPQ CX, $0
-    JNE init
-    RET
-init:
-    ADDQ DI, BX
-    SHRQ $3, CX
     XORQ SI, SI
-
-    // Make sure `size - offset` is at least 4 bytes, otherwise VPGATHERDD
-    // may read data beyond the end of the program memory and trigger a fault.
-    //
-    // If the boolean values do not have enough padding we must fallback to the
-    // scalar algorithm to be able to load single bytes from memory.
-    MOVQ DX, R8
-    SUBQ DI, R8
-    CMPQ R8, $4
-    JB loop
-
-    VPBROADCASTD size+24(FP), Y0
-    VPMULLD ·range0n8(SB), Y0, Y0
-    VPCMPEQD Y1, Y1, Y1
-    VPCMPEQD Y2, Y2, Y2
-    VPCMPEQD Y3, Y3, Y3
-    VPSRLD $31, Y3, Y3
-avx2loop:
-    VPGATHERDD Y1, (BX)(Y0*1), Y4
-    VMOVDQU Y2, Y1
-    VPAND Y3, Y4, Y4
-    VPSLLD $31, Y4, Y4
-    VMOVMSKPS Y4, DI
-
-    MOVB DI, (AX)(SI*1)
-
-    LEAQ (BX)(DX*8), BX
-    INCQ SI
-    CMPQ SI, CX
-    JNE avx2loop
-    VZEROUPPER
-    RET
+    JMP test
 loop:
-    LEAQ (BX)(DX*2), DI
-    MOVBQZX (BX), R8
-    MOVBQZX (BX)(DX*1), R9
-    MOVBQZX (DI), R10
-    MOVBQZX (DI)(DX*1), R11
-    LEAQ (BX)(DX*4), BX
-    LEAQ (DI)(DX*4), DI
-    MOVBQZX (BX), R12
-    MOVBQZX (BX)(DX*1), R13
-    MOVBQZX (DI), R14
-    MOVBQZX (DI)(DX*1), R15
-    LEAQ (BX)(DX*4), BX
-
-    ANDQ $1, R8
-    ANDQ $1, R9
-    ANDQ $1, R10
-    ANDQ $1, R11
-    ANDQ $1, R12
-    ANDQ $1, R13
-    ANDQ $1, R14
-    ANDQ $1, R15
-
-    SHLQ $1, R9
-    SHLQ $2, R10
-    SHLQ $3, R11
-    SHLQ $4, R12
-    SHLQ $5, R13
-    SHLQ $6, R14
-    SHLQ $7, R15
-
-    ORQ R9, R8
-    ORQ R11, R10
-    ORQ R13, R12
-    ORQ R15, R14
-    ORQ R10, R8
-    ORQ R12, R8
-    ORQ R14, R8
-
-    MOVB R8, (AX)(SI*1)
-
-    INCQ SI
-    CMPQ SI, CX
-    JNE loop
-    RET
-
-// func writeValues32bitsAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-TEXT ·writeValues32bitsAVX2(SB), NOSPLIT, $0-40
-    MOVQ values+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), CX
-    MOVQ size+24(FP), DX
-
-    XORQ SI, SI
-    ADDQ offset+32(FP), BX
-
-    CMPQ CX, $0
-    JE done
-
-    CMPQ CX, $8
-    JB loop1x4
-
-    MOVQ CX, DI
-    SHRQ $3, DI
-    SHLQ $3, DI
-
-    VPBROADCASTD size+24(FP), Y0
-    VPMULLD ·range0n8(SB), Y0, Y0
-    VPCMPEQD Y1, Y1, Y1
-    VPCMPEQD Y2, Y2, Y2
-loop8x4:
-    VPGATHERDD Y1, (BX)(Y0*1), Y3
-    VMOVDQU Y3, (AX)(SI*4)
-    VMOVDQU Y2, Y1
-
-    LEAQ (BX)(DX*8), BX
-    ADDQ $8, SI
-    CMPQ SI, DI
-    JNE loop8x4
-    VZEROUPPER
-
-    CMPQ SI, CX
-    JE done
-
-loop1x4:
-    MOVL (BX), R8
-    MOVL R8, (AX)(SI*4)
-
-    ADDQ DX, BX
-    INCQ SI
-    CMPQ SI, CX
-    JNE loop1x4
-done:
-    RET
-
-// func writeValues64bitsAVX2(values unsafe.Pointer, rows array, size, offset uintptr)
-TEXT ·writeValues64bitsAVX2(SB), NOSPLIT, $0-40
-    MOVQ values+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), CX
-    MOVQ size+24(FP), DX
-
-    XORQ SI, SI
-    ADDQ offset+32(FP), BX
-
-    CMPQ CX, $0
-    JE done
-
-    CMPQ CX, $4
-    JB loop1x8
-
-    MOVQ CX, DI
-    SHRQ $2, DI
-    SHLQ $2, DI
-
-    VPBROADCASTQ size+24(FP), Y0
-    VPMULLD ·scale4x8(SB), Y0, Y0
-    VPCMPEQD Y1, Y1, Y1
-    VPCMPEQD Y2, Y2, Y2
-loop4x8:
-    VPGATHERQQ Y1, (BX)(Y0*1), Y3
-    VMOVDQU Y3, (AX)(SI*8)
-    VMOVDQU Y2, Y1
-
-    LEAQ (BX)(DX*4), BX
-    ADDQ $4, SI
-    CMPQ SI, DI
-    JNE loop4x8
-    VZEROUPPER
-
-    CMPQ SI, CX
-    JE done
-loop1x8:
-    MOVQ (BX), R8
-    MOVQ R8, (AX)(SI*8)
-
+    PXOR X0, X0
+    MOVQ (BX), DI // *[16]byte
+    CMPQ DI, $0
+    JE next
+    MOVOU (DI), X0
+next:
+    MOVOU X0, (AX)
+    ADDQ $16, AX
     ADDQ DX, BX
     INCQ SI
+test:
     CMPQ SI, CX
-    JNE loop1x8
-done:
-    RET
-
-// func writeValues128bits(values unsafe.Pointer, rows array, size, offset uintptr)
-TEXT ·writeValues128bits(SB), NOSPLIT, $0-40
-    MOVQ values+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), CX
-    MOVQ size+24(FP), DX
-    ADDQ offset+32(FP), BX
-
-    CMPQ CX, $0
-    JE done
-
-    CMPQ CX, $1
-    JE tail
-
-    XORQ SI, SI
-    MOVQ CX, DI
-    SHRQ $1, DI
-    SHLQ $1, DI
-loop:
-    MOVOU (BX), X0
-    MOVOU (BX)(DX*1), X1
-
-    MOVOU X0, (AX)
-    MOVOU X1, 16(AX)
-
-    LEAQ (BX)(DX*2), BX
-    ADDQ $32, AX
-    ADDQ $2, SI
-    CMPQ SI, DI
     JNE loop
-
-    CMPQ SI, CX
-    JE done
-tail:
-    MOVOU (BX), X0
-    MOVOU X0, (AX)
-done:
     RET
-
-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 5e3eaa3..5f5d7f7 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
 // writeRowsFunc is the type of functions that apply rows to a set of column
@@ -18,15 +19,10 @@ import (
 //
 // - rows is the array of Go values to write to the column buffers.
 //
-// - size is the size of Go values in the rows array (in bytes).
-//
-// - offset is the byte offset of the value being written in each element of the
-//   rows array.
-//
 // - levels is used to track the column index, repetition and definition levels
 //   of values when writing optional or repeated columns.
 //
-type writeRowsFunc func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error
+type writeRowsFunc func(columns []ColumnBuffer, rows sparse.Array, 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
@@ -78,22 +74,22 @@ func writeRowsFuncOf(t reflect.Type, schema *Schema, path columnPath) writeRowsF
 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 func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		columns[columnIndex].writeValues(rows, levels)
 		return nil
 	}
 }
 
 func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, writeRows writeRowsFunc) writeRowsFunc {
 	nullIndex := nullIndexFuncOf(t)
-	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
-		if rows.len == 0 {
-			return writeRows(columns, rows, size, 0, levels)
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
+			return writeRows(columns, rows, levels)
 		}
 
-		nulls := acquireBitmap(rows.len)
+		nulls := acquireBitmap(rows.Len())
 		defer releaseBitmap(nulls)
-		nullIndex(nulls.bits, rows, size, offset)
+		nullIndex(nulls.bits, rows)
 
 		nullLevels := levels
 		levels.definitionLevel++
@@ -113,7 +109,7 @@ func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, wr
 		// This optimization is defeated by inputs alternating null and non-null
 		// sequences of single values, we do not expect this condition to be a
 		// common case.
-		for i := 0; i < rows.len; {
+		for i := 0; i < rows.Len(); {
 			j := 0
 			x := i / 64
 			y := i % 64
@@ -137,13 +133,12 @@ func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, wr
 			}
 
 		writeNulls:
-			if j = x*64 + y; j > rows.len {
-				j = rows.len
+			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, nullLevels); err != nil {
+				if err := writeRows(columns, rows.Slice(i, j), nullLevels); err != nil {
 					return err
 				}
 				i = j
@@ -168,13 +163,12 @@ func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, wr
 			}
 
 		writeNonNulls:
-			if j = x*64 + y; j > rows.len {
-				j = rows.len
+			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 {
+				if err := writeRows(columns, rows.Slice(i, j), levels); err != nil {
 					return err
 				}
 				i = j
@@ -187,7 +181,7 @@ func writeRowsFuncOfOptional(t reflect.Type, schema *Schema, path columnPath, wr
 
 func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	elemType := t.Elem()
-	elemSize := elemType.Size()
+	elemSize := uintptr(elemType.Size())
 	writeRows := writeRowsFuncOf(elemType, schema, path)
 
 	if len(path) == 0 {
@@ -195,19 +189,18 @@ 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(columns []ColumnBuffer, rows array, size, _ uintptr, levels columnLevels) error {
-			if rows.len == 0 {
-				return writeRows(columns, rows, size, 0, levels)
+		return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+			if rows.Len() == 0 {
+				return writeRows(columns, rows, levels)
 			}
 
-			for i := 0; i < rows.len; i++ {
-				p := *(*unsafe.Pointer)(rows.index(i, size, 0))
-				a := array{}
+			for i := 0; i < rows.Len(); i++ {
+				p := *(*unsafe.Pointer)(rows.Index(i))
+				a := sparse.Array{}
 				if p != nil {
-					a.ptr = p
-					a.len = 1
+					a = makeArray(p, 1, elemSize)
 				}
-				if err := writeRows(columns, a, elemSize, 0, levels); err != nil {
+				if err := writeRows(columns, a, levels); err != nil {
 					return err
 				}
 			}
@@ -216,21 +209,20 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 		}
 	}
 
-	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
-		if rows.len == 0 {
-			return writeRows(columns, rows, size, 0, levels)
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
+			return writeRows(columns, rows, levels)
 		}
 
-		for i := 0; i < rows.len; i++ {
-			p := *(*unsafe.Pointer)(rows.index(i, size, offset))
-			a := array{}
+		for i := 0; i < rows.Len(); i++ {
+			p := *(*unsafe.Pointer)(rows.Index(i))
+			a := sparse.Array{}
 			elemLevels := levels
 			if p != nil {
-				a.ptr = p
-				a.len = 1
+				a = makeArray(p, 1, elemSize)
 				elemLevels.definitionLevel++
 			}
-			if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
+			if err := writeRows(columns, a, elemLevels); err != nil {
 				return err
 			}
 		}
@@ -241,36 +233,34 @@ func writeRowsFuncOfPointer(t reflect.Type, schema *Schema, path columnPath) wri
 
 func writeRowsFuncOfSlice(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	elemType := t.Elem()
-	elemSize := elemType.Size()
+	elemSize := uintptr(elemType.Size())
 	writeRows := writeRowsFuncOf(elemType, schema, path)
-	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
-		if rows.len == 0 {
-			return writeRows(columns, rows, size, 0, levels)
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
+			return writeRows(columns, rows, levels)
 		}
 
 		levels.repetitionDepth++
 
-		for i := 0; i < rows.len; i++ {
-			p := rows.index(i, size, offset)
-			a := *(*array)(p)
-			n := a.len
+		for i := 0; i < rows.Len(); i++ {
+			p := (*sliceHeader)(rows.Index(i))
+			a := makeArray(p.base, p.len, elemSize)
+			b := sparse.Array{}
 
 			elemLevels := levels
-			if n > 0 {
-				a.len = 1
+			if a.Len() > 0 {
+				b = a.Slice(0, 1)
 				elemLevels.definitionLevel++
 			}
 
-			if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
+			if err := writeRows(columns, b, elemLevels); err != nil {
 				return err
 			}
 
-			if n > 1 {
+			if a.Len() > 1 {
 				elemLevels.repetitionLevel = elemLevels.repetitionDepth
-				a.ptr = a.index(1, elemSize, 0)
-				a.len = n - 1
 
-				if err := writeRows(columns, a, elemSize, 0, elemLevels); err != nil {
+				if err := writeRows(columns, a.Slice(1, a.Len()), elemLevels); err != nil {
 					return err
 				}
 			}
@@ -316,9 +306,9 @@ func writeRowsFuncOfStruct(t reflect.Type, schema *Schema, path columnPath) writ
 		}
 	}
 
-	return func(buffers []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
+	return func(buffers []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
 		for _, column := range columns {
-			if err := column.writeRows(buffers, rows, size, offset+column.offset, levels); err != nil {
+			if err := column.writeRows(buffers, rows.Offset(column.offset), levels); err != nil {
 				return err
 			}
 		}
@@ -329,26 +319,26 @@ func writeRowsFuncOfStruct(t reflect.Type, schema *Schema, path columnPath) writ
 func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	keyPath := path.append("key_value", "key")
 	keyType := t.Key()
-	keySize := keyType.Size()
+	keySize := uintptr(keyType.Size())
 	writeKeys := writeRowsFuncOf(keyType, schema, keyPath)
 
 	valuePath := path.append("key_value", "value")
 	valueType := t.Elem()
-	valueSize := valueType.Size()
+	valueSize := uintptr(valueType.Size())
 	writeValues := writeRowsFuncOf(valueType, schema, valuePath)
 
-	writeKeyValues := func(columns []ColumnBuffer, keys, values array, levels columnLevels) error {
-		if err := writeKeys(columns, keys, keySize, 0, levels); err != nil {
+	writeKeyValues := func(columns []ColumnBuffer, keys, values sparse.Array, levels columnLevels) error {
+		if err := writeKeys(columns, keys, levels); err != nil {
 			return err
 		}
-		if err := writeValues(columns, values, valueSize, 0, levels); err != nil {
+		if err := writeValues(columns, values, levels); err != nil {
 			return err
 		}
 		return nil
 	}
 
-	return func(columns []ColumnBuffer, rows array, size, offset uintptr, levels columnLevels) error {
-		if rows.len == 0 {
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
 			return writeKeyValues(columns, rows, rows, levels)
 		}
 
@@ -356,11 +346,12 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 		mapKey := reflect.New(keyType).Elem()
 		mapValue := reflect.New(valueType).Elem()
 
-		for i := 0; i < rows.len; i++ {
-			m := reflect.NewAt(t, rows.index(i, size, offset)).Elem()
+		for i := 0; i < rows.Len(); i++ {
+			m := reflect.NewAt(t, rows.Index(i)).Elem()
 
 			if m.Len() == 0 {
-				if err := writeKeyValues(columns, array{}, array{}, levels); err != nil {
+				empty := sparse.Array{}
+				if err := writeKeyValues(columns, empty, empty, levels); err != nil {
 					return err
 				}
 			} else {
@@ -371,8 +362,8 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 					mapKey.SetIterKey(it)
 					mapValue.SetIterValue(it)
 
-					k := makeArray(unsafecast.PointerOfValue(mapKey), 1)
-					v := makeArray(unsafecast.PointerOfValue(mapValue), 1)
+					k := makeArray(unsafecast.PointerOfValue(mapKey), 1, keySize)
+					v := makeArray(unsafecast.PointerOfValue(mapValue), 1, valueSize)
 
 					if err := writeKeyValues(columns, k, v, elemLevels); err != nil {
 						return err
diff --git a/column_buffer_purego.go b/column_buffer_purego.go
index 9db2daf..e0564b0 100644
--- a/column_buffer_purego.go
+++ b/column_buffer_purego.go
@@ -2,6 +2,8 @@
 
 package parquet
 
+import "github.com/segmentio/parquet-go/sparse"
+
 func broadcastValueInt32(dst []int32, src int8) {
 	value := 0x01010101 * int32(src)
 	for i := range dst {
@@ -15,36 +17,14 @@ func broadcastRangeInt32(dst []int32, base int32) {
 	}
 }
 
-func writeValuesBool(values []byte, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesInt32(values []int32, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesInt64(values []int64, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesUint32(values []uint32, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesUint64(values []uint64, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesFloat32(values []float32, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesFloat64(values []float64, rows array, size, offset uintptr) {
-	panic("unreachable")
-}
-
-func writeValuesBE128(values [][16]byte, rows array, size, offset uintptr) {
+func writePointersBE128(values [][16]byte, rows sparse.Array) {
 	for i := range values {
-		values[i] = *(*[16]byte)(rows.index(i, size, offset))
+		p := *(**[16]byte)(rows.Index(i))
+
+		if p != nil {
+			values[i] = *p
+		} else {
+			values[i] = [16]byte{}
+		}
 	}
 }
diff --git a/dictionary.go b/dictionary.go
index 92bc190..d800599 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -8,14 +8,29 @@ import (
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/encoding/plain"
+	"github.com/segmentio/parquet-go/hashprobe"
 	"github.com/segmentio/parquet-go/internal/bitpack"
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
 const (
-	// Completely arbitrary, feel free to adjust if a different value would be
-	// more representative of the map implementation in Go.
-	mapSizeOverheadPerItem = 8
+	// Maximum load of probing tables. This parameter configures the balance
+	// between memory density and compute time of probing operations. Valid
+	// values are floating point numbers between 0 and 1.
+	//
+	// Smaller values result in lower collision probability when inserting
+	// values in probing tables, but also increase memory utilization.
+	//
+	// TODO: make this configurable by the application?
+	hashprobeTableMaxLoad = 0.85
+
+	// An estimate of the CPU cache footprint used by insert operations.
+	//
+	// This constant is used to determine a useful chunk size depending on the
+	// size of values being inserted in dictionaries. More values of small size
+	// can fit in CPU caches, so the inserts can operation on larger chunks.
+	insertsTargetCacheFootprint = 8192
 )
 
 // The Dictionary interface represents type-specific implementations of parquet
@@ -67,12 +82,12 @@ type Dictionary interface {
 
 	// See ColumnBuffer.writeValues for details on the use of unexported methods
 	// on interfaces.
-	insert(indexes []int32, rows array, size, offset uintptr)
-	//lookup(indexes []int32, rows array, size, offset uintptr)
+	insert(indexes []int32, rows sparse.Array)
+	//lookup(indexes []int32, rows sparse.Array)
 }
 
-func checkLookupIndexBounds(indexes []int32, rows array) {
-	if rows.len < len(indexes) {
+func checkLookupIndexBounds(indexes []int32, rows sparse.Array) {
+	if rows.Len() < len(indexes) {
 		panic("dictionary lookup with more indexes than values")
 	}
 }
@@ -86,7 +101,7 @@ type booleanDictionary struct {
 	// dictionary, we ensure that an index exist for each boolean value,
 	// then use the value 0 or 1 (false or true) to perform a lookup in
 	// the dictionary's map.
-	hashmap [2]int32
+	table [2]int32
 }
 
 func newBooleanDictionary(typ Type, columnIndex int16, numValues int32, values []byte) *booleanDictionary {
@@ -109,7 +124,7 @@ func newBooleanDictionary(typ Type, columnIndex int16, numValues int32, values [
 			numValues:   numValues,
 			columnIndex: ^columnIndex,
 		},
-		hashmap: [2]int32{
+		table: [2]int32{
 			0: indexOfFalse,
 			1: indexOfTrue,
 		},
@@ -125,29 +140,29 @@ func (d *booleanDictionary) Index(i int32) Value { return d.makeValue(d.index(i)
 func (d *booleanDictionary) index(i int32) bool { return d.valueAt(int(i)) }
 
 func (d *booleanDictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *booleanDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *booleanDictionary) insert(indexes []int32, rows sparse.Array) {
+	_ = indexes[:rows.Len()]
 
-	if d.hashmap[0] < 0 {
-		d.hashmap[0] = d.numValues
+	if d.table[0] < 0 {
+		d.table[0] = d.numValues
 		d.numValues++
-		d.bits = plain.AppendBoolean(d.bits, int(d.hashmap[0]), false)
+		d.bits = plain.AppendBoolean(d.bits, int(d.table[0]), false)
 	}
 
-	if d.hashmap[1] < 0 {
-		d.hashmap[1] = d.numValues
+	if d.table[1] < 0 {
+		d.table[1] = d.numValues
 		d.numValues++
-		d.bits = plain.AppendBoolean(d.bits, int(d.hashmap[1]), true)
+		d.bits = plain.AppendBoolean(d.bits, int(d.table[1]), true)
 	}
 
-	dict := d.hashmap
+	dict := d.table
 
-	for i := 0; i < rows.len; i++ {
-		v := *(*byte)(rows.index(i, size, offset)) & 1
+	for i := 0; i < rows.Len(); i++ {
+		v := *(*byte)(rows.Index(i)) & 1
 		indexes[i] = dict[v]
 	}
 }
@@ -155,13 +170,13 @@ func (d *booleanDictionary) insert(indexes []int32, rows array, size, offset uin
 func (d *booleanDictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(false)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *booleanDictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *booleanDictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*bool)(rows.index(i, size, offset)) = d.index(j)
+		*(*bool)(rows.Index(i)) = d.index(j)
 	}
 }
 
@@ -191,7 +206,7 @@ func (d *booleanDictionary) Reset() {
 	d.bits = d.bits[:0]
 	d.offset = 0
 	d.numValues = 0
-	d.hashmap = [2]int32{-1, -1}
+	d.table = [2]int32{-1, -1}
 }
 
 func (d *booleanDictionary) Page() BufferedPage {
@@ -200,7 +215,7 @@ func (d *booleanDictionary) Page() BufferedPage {
 
 type int32Dictionary struct {
 	int32Page
-	hashmap map[int32]int32
+	table *hashprobe.Int32Table
 }
 
 func newInt32Dictionary(typ Type, columnIndex int16, numValues int32, values []byte) *int32Dictionary {
@@ -222,38 +237,60 @@ func (d *int32Dictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *int32Dictionary) index(i int32) int32 { return d.values[i] }
 
 func (d *int32Dictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *int32Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *int32Dictionary) init(indexes []int32) {
+	d.table = hashprobe.NewInt32Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[int32]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
+	}
+}
+
+func (d *int32Dictionary) insert(indexes []int32, rows sparse.Array) {
+	// Iterating over the input in chunks helps keep relevant data in CPU
+	// caches when a large number of values are inserted into the dictionary with
+	// a single method call.
+	//
+	// Without this chunking, memory areas from the head of the indexes and
+	// values arrays end up being evicted from CPU caches as the probing
+	// operation iterates through the array. The subsequent scan of the indexes
+	// required to determine which values must be inserted into the page then
+	// stalls on retrieving data from main memory.
+	//
+	// We measured as much as ~37% drop in throughput when disabling the
+	// chunking, and did not observe any penalties from having it on smaller
+	// inserts.
+	const chunkSize = insertsTargetCacheFootprint / 4
+
+	if d.table == nil {
+		d.init(indexes)
 	}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*int32)(rows.index(i, size, offset))
+	values := rows.Int32Array()
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
 
-		indexes[i] = index
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *int32Dictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *int32Dictionary) Bounds(indexes []int32) (min, max Value) {
@@ -267,7 +304,9 @@ func (d *int32Dictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *int32Dictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *int32Dictionary) Page() BufferedPage {
@@ -276,7 +315,7 @@ func (d *int32Dictionary) Page() BufferedPage {
 
 type int64Dictionary struct {
 	int64Page
-	hashmap map[int64]int32
+	table *hashprobe.Int64Table
 }
 
 func newInt64Dictionary(typ Type, columnIndex int16, numValues int32, values []byte) *int64Dictionary {
@@ -298,38 +337,47 @@ func (d *int64Dictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *int64Dictionary) index(i int32) int64 { return d.values[i] }
 
 func (d *int64Dictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *int64Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *int64Dictionary) init(indexes []int32) {
+	d.table = hashprobe.NewInt64Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[int64]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
 	}
+}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*int64)(rows.index(i, size, offset))
+func (d *int64Dictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 8
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	if d.table == nil {
+		d.init(indexes)
+	}
 
-		indexes[i] = index
+	values := rows.Int64Array()
+
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
+
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *int64Dictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *int64Dictionary) Bounds(indexes []int32) (min, max Value) {
@@ -343,7 +391,9 @@ func (d *int64Dictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *int64Dictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *int64Dictionary) Page() BufferedPage {
@@ -379,9 +429,9 @@ func (d *int96Dictionary) Insert(indexes []int32, values []Value) {
 	})
 }
 
-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) insert(indexes []int32, rows sparse.Array) {
+	d.insertValues(indexes, rows.Len(), func(i int) deprecated.Int96 {
+		return *(*deprecated.Int96)(rows.Index(i))
 	})
 }
 
@@ -447,7 +497,7 @@ func (d *int96Dictionary) Page() BufferedPage {
 
 type floatDictionary struct {
 	floatPage
-	hashmap map[float32]int32
+	table *hashprobe.Float32Table
 }
 
 func newFloatDictionary(typ Type, columnIndex int16, numValues int32, values []byte) *floatDictionary {
@@ -469,38 +519,47 @@ func (d *floatDictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *floatDictionary) index(i int32) float32 { return d.values[i] }
 
 func (d *floatDictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *floatDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *floatDictionary) init(indexes []int32) {
+	d.table = hashprobe.NewFloat32Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[float32]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
 	}
+}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*float32)(rows.index(i, size, offset))
+func (d *floatDictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 4
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	if d.table == nil {
+		d.init(indexes)
+	}
 
-		indexes[i] = index
+	values := rows.Float32Array()
+
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
+
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *floatDictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *floatDictionary) Bounds(indexes []int32) (min, max Value) {
@@ -514,7 +573,9 @@ func (d *floatDictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *floatDictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *floatDictionary) Page() BufferedPage {
@@ -523,7 +584,7 @@ func (d *floatDictionary) Page() BufferedPage {
 
 type doubleDictionary struct {
 	doublePage
-	hashmap map[float64]int32
+	table *hashprobe.Float64Table
 }
 
 func newDoubleDictionary(typ Type, columnIndex int16, numValues int32, values []byte) *doubleDictionary {
@@ -545,38 +606,47 @@ func (d *doubleDictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *doubleDictionary) index(i int32) float64 { return d.values[i] }
 
 func (d *doubleDictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *doubleDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *doubleDictionary) init(indexes []int32) {
+	d.table = hashprobe.NewFloat64Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[float64]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
 	}
+}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*float64)(rows.index(i, size, offset))
+func (d *doubleDictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 8
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	if d.table == nil {
+		d.init(indexes)
+	}
 
-		indexes[i] = index
+	values := rows.Float64Array()
+
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
+
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *doubleDictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *doubleDictionary) Bounds(indexes []int32) (min, max Value) {
@@ -590,7 +660,9 @@ func (d *doubleDictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *doubleDictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *doubleDictionary) Page() BufferedPage {
@@ -633,12 +705,12 @@ func (d *byteArrayDictionary) Index(i int32) Value { return d.makeValueBytes(d.i
 func (d *byteArrayDictionary) index(i int32) []byte { return d.valueAt(d.offsets[i]) }
 
 func (d *byteArrayDictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.ptr))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.ptr)))
 }
 
-func (d *byteArrayDictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *byteArrayDictionary) insert(indexes []int32, rows sparse.Array) {
+	_ = indexes[:rows.Len()]
 
 	if d.hashmap == nil {
 		d.hashmap = make(map[string]int32, cap(d.offsets))
@@ -647,8 +719,8 @@ func (d *byteArrayDictionary) insert(indexes []int32, rows array, size, offset u
 		}
 	}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*string)(rows.index(i, size, offset))
+	for i := 0; i < rows.Len(); i++ {
+		value := *(*string)(rows.Index(i))
 
 		index, exists := d.hashmap[value]
 		if !exists {
@@ -672,7 +744,7 @@ func (d *byteArrayDictionary) append(value string) string {
 func (d *byteArrayDictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValueString("")
 	memsetValues(values, model)
-	d.lookupString(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.ptr))
+	d.lookupString(indexes, makeArrayValue(values, unsafe.Offsetof(model.ptr)))
 }
 
 func (d *byteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
@@ -688,7 +760,7 @@ func (d *byteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
 				n = len(values)
 			}
 			j := i + n
-			d.lookupString(indexes[i:j:j], makeArrayString(values[:n:n]), unsafe.Sizeof(values[0]), 0)
+			d.lookupString(indexes[i:j:j], makeArrayString(values[:n:n]))
 
 			for _, value := range values[:n:n] {
 				switch {
@@ -754,9 +826,9 @@ func (d *fixedLenByteArrayDictionary) Insert(indexes []int32, values []Value) {
 	})
 }
 
-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) insert(indexes []int32, rows sparse.Array) {
+	d.insertValues(indexes, rows.Len(), func(i int) *byte {
+		return (*byte)(rows.Index(i))
 	})
 }
 
@@ -789,7 +861,7 @@ func (d *fixedLenByteArrayDictionary) insertValues(indexes []int32, count int, v
 func (d *fixedLenByteArrayDictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValueString("")
 	memsetValues(values, model)
-	d.lookupString(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.ptr))
+	d.lookupString(indexes, makeArrayValue(values, unsafe.Offsetof(model.ptr)))
 }
 
 func (d *fixedLenByteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
@@ -805,7 +877,7 @@ func (d *fixedLenByteArrayDictionary) Bounds(indexes []int32) (min, max Value) {
 				n = len(values)
 			}
 			j := i + n
-			d.lookupString(indexes[i:j:j], makeArrayString(values[:n:n]), unsafe.Sizeof(values[0]), 0)
+			d.lookupString(indexes[i:j:j], makeArrayString(values[:n:n]))
 
 			for _, value := range values[:n:n] {
 				switch {
@@ -834,7 +906,7 @@ func (d *fixedLenByteArrayDictionary) Page() BufferedPage {
 
 type uint32Dictionary struct {
 	uint32Page
-	hashmap map[uint32]int32
+	table *hashprobe.Uint32Table
 }
 
 func newUint32Dictionary(typ Type, columnIndex int16, numValues int32, data []byte) *uint32Dictionary {
@@ -856,38 +928,47 @@ func (d *uint32Dictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *uint32Dictionary) index(i int32) uint32 { return d.values[i] }
 
 func (d *uint32Dictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *uint32Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *uint32Dictionary) init(indexes []int32) {
+	d.table = hashprobe.NewUint32Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[uint32]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
+	}
+}
+
+func (d *uint32Dictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 4
+
+	if d.table == nil {
+		d.init(indexes)
 	}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*uint32)(rows.index(i, size, offset))
+	values := rows.Uint32Array()
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
 
-		indexes[i] = index
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *uint32Dictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *uint32Dictionary) Bounds(indexes []int32) (min, max Value) {
@@ -901,7 +982,9 @@ func (d *uint32Dictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *uint32Dictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *uint32Dictionary) Page() BufferedPage {
@@ -910,7 +993,7 @@ func (d *uint32Dictionary) Page() BufferedPage {
 
 type uint64Dictionary struct {
 	uint64Page
-	hashmap map[uint64]int32
+	table *hashprobe.Uint64Table
 }
 
 func newUint64Dictionary(typ Type, columnIndex int16, numValues int32, data []byte) *uint64Dictionary {
@@ -932,38 +1015,47 @@ func (d *uint64Dictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *uint64Dictionary) index(i int32) uint64 { return d.values[i] }
 
 func (d *uint64Dictionary) Insert(indexes []int32, values []Value) {
-	var value Value
-	d.insert(indexes, makeArrayValue(values), unsafe.Sizeof(value), unsafe.Offsetof(value.u64))
+	model := Value{}
+	d.insert(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
-func (d *uint64Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	_ = indexes[:rows.len]
+func (d *uint64Dictionary) init(indexes []int32) {
+	d.table = hashprobe.NewUint64Table(cap(d.values), hashprobeTableMaxLoad)
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[uint64]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
 	}
+}
 
-	for i := 0; i < rows.len; i++ {
-		value := *(*uint64)(rows.index(i, size, offset))
+func (d *uint64Dictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 8
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	if d.table == nil {
+		d.init(indexes)
+	}
 
-		indexes[i] = index
+	values := rows.Uint64Array()
+
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
+
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *uint64Dictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValue(0)
 	memsetValues(values, model)
-	d.lookup(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.u64))
+	d.lookup(indexes, makeArrayValue(values, unsafe.Offsetof(model.u64)))
 }
 
 func (d *uint64Dictionary) Bounds(indexes []int32) (min, max Value) {
@@ -977,7 +1069,9 @@ func (d *uint64Dictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *uint64Dictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *uint64Dictionary) Page() BufferedPage {
@@ -986,7 +1080,7 @@ func (d *uint64Dictionary) Page() BufferedPage {
 
 type be128Dictionary struct {
 	be128Page
-	hashmap map[[16]byte]int32
+	table *hashprobe.Uint128Table
 }
 
 func newBE128Dictionary(typ Type, columnIndex int16, numValues int32, data []byte) *be128Dictionary {
@@ -1008,45 +1102,78 @@ func (d *be128Dictionary) Index(i int32) Value { return d.makeValue(d.index(i))
 func (d *be128Dictionary) index(i int32) *[16]byte { return &d.values[i] }
 
 func (d *be128Dictionary) Insert(indexes []int32, values []Value) {
-	d.insertValues(indexes, len(values), func(i int) [16]byte {
-		return *(*[16]byte)(values[i].ByteArray())
-	})
+	_ = indexes[:len(values)]
+
+	for _, v := range values {
+		if v.kind != ^int8(FixedLenByteArray) {
+			panic("values inserted in BE128 dictionary must be of type BYTE_ARRAY")
+		}
+		if v.u64 != 16 {
+			panic("values inserted in BE128 dictionary must be of length 16")
+		}
+	}
+
+	if d.table == nil {
+		d.init(indexes)
+	}
+
+	const chunkSize = insertsTargetCacheFootprint / 16
+	var buffer [chunkSize][16]byte
+
+	for i := 0; i < len(values); i += chunkSize {
+		j := min(chunkSize+i, len(values))
+		n := min(chunkSize, len(values)-i)
+
+		probe := buffer[:n:n]
+		writePointersBE128(probe, makeArrayValue(values[i:j], unsafe.Offsetof(values[i].ptr)))
+
+		if d.table.Probe(probe, indexes[i:j:j]) > 0 {
+			for k, v := range probe {
+				if indexes[i+k] == int32(len(d.values)) {
+					d.values = append(d.values, v)
+				}
+			}
+		}
+	}
 }
 
-func (d *be128Dictionary) insert(indexes []int32, rows array, size, offset uintptr) {
-	d.insertValues(indexes, rows.len, func(i int) [16]byte {
-		return *(*[16]byte)(rows.index(i, size, offset))
-	})
+func (d *be128Dictionary) init(indexes []int32) {
+	d.table = hashprobe.NewUint128Table(cap(d.values), 0.75)
+
+	n := min(len(d.values), len(indexes))
+
+	for i := 0; i < len(d.values); i += n {
+		j := min(i+n, len(d.values))
+		d.table.Probe(d.values[i:j:j], indexes[:n:n])
+	}
 }
 
-func (d *be128Dictionary) insertValues(indexes []int32, count int, valueAt func(int) [16]byte) {
-	_ = indexes[:count]
+func (d *be128Dictionary) insert(indexes []int32, rows sparse.Array) {
+	const chunkSize = insertsTargetCacheFootprint / 16
 
-	if d.hashmap == nil {
-		d.hashmap = make(map[[16]byte]int32, cap(d.values))
-		for i, v := range d.values {
-			d.hashmap[v] = int32(i)
-		}
+	if d.table == nil {
+		d.init(indexes)
 	}
 
-	for i := 0; i < count; i++ {
-		value := valueAt(i)
+	values := rows.Uint128Array()
 
-		index, exists := d.hashmap[value]
-		if !exists {
-			index = int32(len(d.values))
-			d.values = append(d.values, value)
-			d.hashmap[value] = index
-		}
+	for i := 0; i < values.Len(); i += chunkSize {
+		j := min(i+chunkSize, values.Len())
 
-		indexes[i] = index
+		if d.table.ProbeArray(values.Slice(i, j), indexes[i:j:j]) > 0 {
+			for k, index := range indexes[i:j] {
+				if index == int32(len(d.values)) {
+					d.values = append(d.values, values.Index(i+k))
+				}
+			}
+		}
 	}
 }
 
 func (d *be128Dictionary) Lookup(indexes []int32, values []Value) {
 	model := d.makeValueString("")
 	memsetValues(values, model)
-	d.lookupString(indexes, makeArrayValue(values), unsafe.Sizeof(model), unsafe.Offsetof(model.ptr))
+	d.lookupString(indexes, makeArrayValue(values, unsafe.Offsetof(model.ptr)))
 }
 
 func (d *be128Dictionary) Bounds(indexes []int32) (min, max Value) {
@@ -1060,7 +1187,9 @@ func (d *be128Dictionary) Bounds(indexes []int32) (min, max Value) {
 
 func (d *be128Dictionary) Reset() {
 	d.values = d.values[:0]
-	d.hashmap = nil
+	if d.table != nil {
+		d.table.Reset()
+	}
 }
 
 func (d *be128Dictionary) Page() BufferedPage {
@@ -1274,9 +1403,9 @@ func (col *indexedColumnBuffer) WriteValues(values []Value) (int, error) {
 	return len(values), nil
 }
 
-func (col *indexedColumnBuffer) writeValues(rows array, size, offset uintptr, _ columnLevels) {
+func (col *indexedColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
 	i := len(col.values)
-	j := len(col.values) + rows.len
+	j := len(col.values) + rows.Len()
 
 	if j <= cap(col.values) {
 		col.values = col.values[:j]
@@ -1286,7 +1415,7 @@ func (col *indexedColumnBuffer) writeValues(rows array, size, offset uintptr, _
 		col.values = tmp
 	}
 
-	col.typ.dict.insert(col.values[i:], rows, size, offset)
+	col.typ.dict.insert(col.values[i:], rows)
 }
 
 func (col *indexedColumnBuffer) ReadValuesAt(values []Value, offset int64) (n int, err error) {
diff --git a/dictionary_amd64.go b/dictionary_amd64.go
index d639a31..33d2ab8 100644
--- a/dictionary_amd64.go
+++ b/dictionary_amd64.go
@@ -4,6 +4,7 @@ package parquet
 
 import (
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
 //go:noescape
@@ -28,74 +29,74 @@ func dictionaryBoundsUint64(dict []uint64, indexes []int32) (min, max uint64, er
 func dictionaryBoundsBE128(dict [][16]byte, indexes []int32) (min, max *[16]byte, err errno)
 
 //go:noescape
-func dictionaryLookup32bits(dict []uint32, indexes []int32, rows array, size, offset uintptr) errno
+func dictionaryLookup32(dict []uint32, indexes []int32, rows sparse.Array) errno
 
 //go:noescape
-func dictionaryLookup64bits(dict []uint64, indexes []int32, rows array, size, offset uintptr) errno
+func dictionaryLookup64(dict []uint64, indexes []int32, rows sparse.Array) errno
 
 //go:noescape
-func dictionaryLookupByteArrayString(dict []uint32, page []byte, indexes []int32, rows array, size, offset uintptr) errno
+func dictionaryLookupByteArrayString(dict []uint32, page []byte, indexes []int32, rows sparse.Array) errno
 
 //go:noescape
-func dictionaryLookupFixedLenByteArrayString(dict []byte, len int, indexes []int32, rows array, size, offset uintptr) errno
+func dictionaryLookupFixedLenByteArrayString(dict []byte, len int, indexes []int32, rows sparse.Array) errno
 
 //go:noescape
-func dictionaryLookupFixedLenByteArrayPointer(dict []byte, len int, indexes []int32, rows array, size, offset uintptr) errno
+func dictionaryLookupFixedLenByteArrayPointer(dict []byte, len int, indexes []int32, rows sparse.Array) errno
 
-func (d *int32Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *int32Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Int32ToUint32(d.values)
-	dictionaryLookup32bits(dict, indexes, rows, size, offset).check()
+	dictionaryLookup32(dict, indexes, rows).check()
 }
 
-func (d *int64Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *int64Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Int64ToUint64(d.values)
-	dictionaryLookup64bits(dict, indexes, rows, size, offset).check()
+	dictionaryLookup64(dict, indexes, rows).check()
 }
 
-func (d *floatDictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *floatDictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Float32ToUint32(d.values)
-	dictionaryLookup32bits(dict, indexes, rows, size, offset).check()
+	dictionaryLookup32(dict, indexes, rows).check()
 }
 
-func (d *doubleDictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *doubleDictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Float64ToUint64(d.values)
-	dictionaryLookup64bits(dict, indexes, rows, size, offset).check()
+	dictionaryLookup64(dict, indexes, rows).check()
 }
 
-func (d *byteArrayDictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *byteArrayDictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
-	dictionaryLookupByteArrayString(d.offsets, d.values, indexes, rows, size, offset).check()
+	dictionaryLookupByteArrayString(d.offsets, d.values, indexes, rows).check()
 }
 
-func (d *fixedLenByteArrayDictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *fixedLenByteArrayDictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
-	dictionaryLookupFixedLenByteArrayString(d.data, d.size, indexes, rows, size, offset).check()
+	dictionaryLookupFixedLenByteArrayString(d.data, d.size, indexes, rows).check()
 }
 
-func (d *uint32Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *uint32Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
-	dictionaryLookup32bits(d.values, indexes, rows, size, offset).check()
+	dictionaryLookup32(d.values, indexes, rows).check()
 }
 
-func (d *uint64Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *uint64Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
-	dictionaryLookup64bits(d.values, indexes, rows, size, offset).check()
+	dictionaryLookup64(d.values, indexes, rows).check()
 }
 
-func (d *be128Dictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *be128Dictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Uint128ToBytes(d.values)
-	dictionaryLookupFixedLenByteArrayString(dict, 16, indexes, rows, size, offset).check()
+	dictionaryLookupFixedLenByteArrayString(dict, 16, indexes, rows).check()
 }
 
-func (d *be128Dictionary) lookupPointer(indexes []int32, rows array, size, offset uintptr) {
+func (d *be128Dictionary) lookupPointer(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	dict := unsafecast.Uint128ToBytes(d.values)
-	dictionaryLookupFixedLenByteArrayPointer(dict, 16, indexes, rows, size, offset).check()
+	dictionaryLookupFixedLenByteArrayPointer(dict, 16, indexes, rows).check()
 }
 
 func (d *int32Dictionary) bounds(indexes []int32) (min, max int32) {
diff --git a/dictionary_amd64.s b/dictionary_amd64.s
index af92e4d..0cd09de 100644
--- a/dictionary_amd64.s
+++ b/dictionary_amd64.s
@@ -598,10 +598,8 @@ TEXT ·dictionaryBoundsBE128(SB), NOSPLIT, $0-72
     SHLQ $4, DI // the dictionary contains 16 byte words
     LEAQ (AX)(DI*1), R8
     MOVQ R8, R9
-    MOVQ 0(AX)(DI*1), R10 // min (high)
-    MOVQ 8(AX)(DI*1), R11 // min (low)
-    BSWAPQ R10
-    BSWAPQ R11
+    MOVBEQQ 0(AX)(DI*1), R10 // min (high)
+    MOVBEQQ 8(AX)(DI*1), R11 // min (low)
     MOVQ R10, R12 // max (high)
     MOVQ R11, R13 // max (low)
 
@@ -657,17 +655,16 @@ indexOutOfBounds:
 // values in the dictionary, then VPSCATTER* to do 8 parallel writes to the
 // sparse output buffer.
 
-// func dictionaryLookup32bits(dict []uint32, indexes []int32, rows array, size, offset uintptr) errno
-TEXT ·dictionaryLookup32bits(SB), NOSPLIT, $0-88
+// func dictionaryLookup32(dict []uint32, indexes []int32, rows sparse.Array) errno
+TEXT ·dictionaryLookup32(SB), NOSPLIT, $0-80
     MOVQ dict_base+0(FP), AX
     MOVQ dict_len+8(FP), BX
 
     MOVQ indexes_base+24(FP), CX
     MOVQ indexes_len+32(FP), DX
 
-    MOVQ rows_ptr+48(FP), R8
-    MOVQ size+64(FP), R9
-    ADDQ offset+72(FP), R8
+    MOVQ rows_array_ptr+48(FP), R8
+    MOVQ rows_array_off+64(FP), R9
 
     XORQ SI, SI
 
@@ -720,23 +717,22 @@ test:
     JNE loop
     XORQ AX, AX
 return:
-    MOVQ AX, ret+80(FP)
+    MOVQ AX, ret+72(FP)
     RET
 indexOutOfBounds:
     MOVQ $errnoIndexOutOfBounds, AX
     JMP return
 
-// func dictionaryLookup64bits(dict []uint64, indexes []int32, rows array, size, offset uintptr) errno
-TEXT ·dictionaryLookup64bits(SB), NOSPLIT, $0-88
+// func dictionaryLookup64(dict []uint64, indexes []int32, rows sparse.Array) errno
+TEXT ·dictionaryLookup64(SB), NOSPLIT, $0-80
     MOVQ dict_base+0(FP), AX
     MOVQ dict_len+8(FP), BX
 
     MOVQ indexes_base+24(FP), CX
     MOVQ indexes_len+32(FP), DX
 
-    MOVQ rows_ptr+48(FP), R8
-    MOVQ size+64(FP), R9
-    ADDQ offset+72(FP), R8
+    MOVQ rows_array_ptr+48(FP), R8
+    MOVQ rows_array_off+64(FP), R9
 
     XORQ SI, SI
 
@@ -789,14 +785,14 @@ test:
     JNE loop
     XORQ AX, AX
 return:
-    MOVQ AX, ret+80(FP)
+    MOVQ AX, ret+72(FP)
     RET
 indexOutOfBounds:
     MOVQ $errnoIndexOutOfBounds, AX
     JMP return
 
-// func dictionaryLookupByteArrayString(dict []uint32, page []byte, indexes []int32, rows array, size, offset uintptr) errno
-TEXT ·dictionaryLookupByteArrayString(SB), NOSPLIT, $0-112
+// func dictionaryLookupByteArrayString(dict []uint32, page []byte, indexes []int32, rows sparse.Array) errno
+TEXT ·dictionaryLookupByteArrayString(SB), NOSPLIT, $0-104
     MOVQ dict_base+0(FP), AX
     MOVQ dict_len+8(FP), BX
 
@@ -805,9 +801,8 @@ TEXT ·dictionaryLookupByteArrayString(SB), NOSPLIT, $0-112
     MOVQ indexes_base+48(FP), R8
     MOVQ indexes_len+56(FP), R9
 
-    MOVQ rows_ptr+72(FP), R10
-    MOVQ size+88(FP), R11
-    ADDQ offset+96(FP), R10
+    MOVQ rows_array_ptr+72(FP), R10
+    MOVQ rows_array_off+88(FP), R11
 
     XORQ DI, DI
     XORQ SI, SI
@@ -844,14 +839,14 @@ test:
     JNE loop
     XORQ AX, AX
 return:
-    MOVQ AX, ret+104(FP)
+    MOVQ AX, ret+96(FP)
     RET
 indexOutOfBounds:
     MOVQ $errnoIndexOutOfBounds, AX
     JMP return
 
-// func dictionaryLookupFixedLenByteArrayString(dict []byte, len int, indexes []int32, rows array, size, offset uintptr) errno
-TEXT ·dictionaryLookupFixedLenByteArrayString(SB), NOSPLIT, $0-96
+// func dictionaryLookupFixedLenByteArrayString(dict []byte, len int, indexes []int32, rows sparse.Array) errno
+TEXT ·dictionaryLookupFixedLenByteArrayString(SB), NOSPLIT, $0-88
     MOVQ dict_base+0(FP), AX
     MOVQ dict_len+8(FP), BX
 
@@ -860,9 +855,8 @@ TEXT ·dictionaryLookupFixedLenByteArrayString(SB), NOSPLIT, $0-96
     MOVQ indexes_base+32(FP), DX
     MOVQ indexes_len+40(FP), R8
 
-    MOVQ rows_ptr+56(FP), R9
-    MOVQ size+72(FP), R10
-    ADDQ offset+80(FP), R9
+    MOVQ rows_array_ptr+56(FP), R9
+    MOVQ rows_array_off+72(FP), R10
 
     XORQ DI, DI
     XORQ SI, SI
@@ -883,7 +877,7 @@ test:
     JNE loop
     XORQ AX, AX
 return:
-    MOVQ AX, ret+88(FP)
+    MOVQ AX, ret+80(FP)
     RET
 indexOutOfBounds:
     MOVQ $errnoIndexOutOfBounds, AX
@@ -894,8 +888,8 @@ indexOutOfBounds:
 // the pair of pointer and length. Since the length is fixed for this dictionary
 // type, the application can assume it at the call site.
 //
-// func dictionaryLookupFixedLenByteArrayPointer(dict []byte, len int, indexes []int32, rows array, size, offset uintptr) errno
-TEXT ·dictionaryLookupFixedLenByteArrayPointer(SB), NOSPLIT, $0-96
+// func dictionaryLookupFixedLenByteArrayPointer(dict []byte, len int, indexes []int32, rows sparse.Array) errno
+TEXT ·dictionaryLookupFixedLenByteArrayPointer(SB), NOSPLIT, $0-88
     MOVQ dict_base+0(FP), AX
     MOVQ dict_len+8(FP), BX
 
@@ -904,9 +898,8 @@ TEXT ·dictionaryLookupFixedLenByteArrayPointer(SB), NOSPLIT, $0-96
     MOVQ indexes_base+32(FP), DX
     MOVQ indexes_len+40(FP), R8
 
-    MOVQ rows_ptr+56(FP), R9
-    MOVQ size+72(FP), R10
-    ADDQ offset+80(FP), R9
+    MOVQ rows_array_ptr+56(FP), R9
+    MOVQ rows_array_off+72(FP), R10
 
     XORQ DI, DI
     XORQ SI, SI
@@ -926,7 +919,7 @@ test:
     JNE loop
     XORQ AX, AX
 return:
-    MOVQ AX, ret+88(FP)
+    MOVQ AX, ret+80(FP)
     RET
 indexOutOfBounds:
     MOVQ $errnoIndexOutOfBounds, AX
diff --git a/dictionary_purego.go b/dictionary_purego.go
index c778a89..2368886 100644
--- a/dictionary_purego.go
+++ b/dictionary_purego.go
@@ -2,79 +2,83 @@
 
 package parquet
 
-import "unsafe"
+import (
+	"unsafe"
 
-func (d *int32Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+	"github.com/segmentio/parquet-go/sparse"
+)
+
+func (d *int32Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*int32)(rows.index(i, size, offset)) = d.index(j)
+		*(*int32)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *int64Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *int64Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*int64)(rows.index(i, size, offset)) = d.index(j)
+		*(*int64)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *floatDictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *floatDictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*float32)(rows.index(i, size, offset)) = d.index(j)
+		*(*float32)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *doubleDictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *doubleDictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*float64)(rows.index(i, size, offset)) = d.index(j)
+		*(*float64)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *byteArrayDictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *byteArrayDictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
 		v := d.index(j)
-		*(*string)(rows.index(i, size, offset)) = *(*string)(unsafe.Pointer(&v))
+		*(*string)(rows.Index(i)) = *(*string)(unsafe.Pointer(&v))
 	}
 }
 
-func (d *fixedLenByteArrayDictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *fixedLenByteArrayDictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
 		v := d.index(j)
-		*(*string)(rows.index(i, size, offset)) = *(*string)(unsafe.Pointer(&v))
+		*(*string)(rows.Index(i)) = *(*string)(unsafe.Pointer(&v))
 	}
 }
 
-func (d *uint32Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *uint32Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*uint32)(rows.index(i, size, offset)) = d.index(j)
+		*(*uint32)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *uint64Dictionary) lookup(indexes []int32, rows array, size, offset uintptr) {
+func (d *uint64Dictionary) lookup(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(*uint64)(rows.index(i, size, offset)) = d.index(j)
+		*(*uint64)(rows.Index(i)) = d.index(j)
 	}
 }
 
-func (d *be128Dictionary) lookupString(indexes []int32, rows array, size, offset uintptr) {
+func (d *be128Dictionary) lookupString(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	s := "0123456789ABCDEF"
 	for i, j := range indexes {
 		*(**[16]byte)(unsafe.Pointer(&s)) = d.index(j)
-		*(*string)(rows.index(i, size, offset)) = s
+		*(*string)(rows.Index(i)) = s
 	}
 }
 
-func (d *be128Dictionary) lookupPointer(indexes []int32, rows array, size, offset uintptr) {
+func (d *be128Dictionary) lookupPointer(indexes []int32, rows sparse.Array) {
 	checkLookupIndexBounds(indexes, rows)
 	for i, j := range indexes {
-		*(**[16]byte)(rows.index(i, size, offset)) = d.index(j)
+		*(**[16]byte)(rows.Index(i)) = d.index(j)
 	}
 }
 
@@ -191,7 +195,7 @@ func (d *be128Dictionary) bounds(indexes []int32) (min, max *[16]byte) {
 			n = len(values)
 		}
 		j := i + n
-		d.lookupPointer(indexes[i:j:j], makeArrayBE128(values[:n:n]), unsafe.Sizeof(values[0]), 0)
+		d.lookupPointer(indexes[i:j:j], makeArrayBE128(values[:n:n]))
 
 		for _, value := range values[:n:n] {
 			switch {
diff --git a/encoding/rle/rle.go b/encoding/rle/rle.go
index 968c0ef..1aa03c3 100644
--- a/encoding/rle/rle.go
+++ b/encoding/rle/rle.go
@@ -506,51 +506,3 @@ func encodeInt32BitpackDefault(dst []byte, src [][8]int32, bitWidth uint) int {
 	bitpack.PackInt32(dst, bits, bitWidth)
 	return bitpack.ByteCount(uint(len(src)*8) * bitWidth)
 }
-
-func encodeBytesBitpackDefault(dst []byte, src []uint64, bitWidth uint) int {
-	bitMask := uint64(1<<bitWidth) - 1
-	n := 0
-
-	for _, word := range src {
-		word = (word & bitMask) |
-			(((word >> 8) & bitMask) << (1 * bitWidth)) |
-			(((word >> 16) & bitMask) << (2 * bitWidth)) |
-			(((word >> 24) & bitMask) << (3 * bitWidth)) |
-			(((word >> 32) & bitMask) << (4 * bitWidth)) |
-			(((word >> 40) & bitMask) << (5 * bitWidth)) |
-			(((word >> 48) & bitMask) << (6 * bitWidth)) |
-			(((word >> 56) & bitMask) << (7 * bitWidth))
-		binary.LittleEndian.PutUint64(dst[n:], word)
-		n += int(bitWidth)
-	}
-
-	return n
-}
-
-func decodeBytesBitpackDefault(dst, src []byte, count, bitWidth uint) {
-	dst = dst[:0]
-
-	bitMask := uint64(1<<bitWidth) - 1
-	byteCount := bitpack.ByteCount(8 * bitWidth)
-
-	for i := 0; count > 0; count -= 8 {
-		j := i + byteCount
-
-		bits := [8]byte{}
-		copy(bits[:], src[i:j])
-		word := binary.LittleEndian.Uint64(bits[:])
-
-		dst = append(dst,
-			byte((word>>(0*bitWidth))&bitMask),
-			byte((word>>(1*bitWidth))&bitMask),
-			byte((word>>(2*bitWidth))&bitMask),
-			byte((word>>(3*bitWidth))&bitMask),
-			byte((word>>(4*bitWidth))&bitMask),
-			byte((word>>(5*bitWidth))&bitMask),
-			byte((word>>(6*bitWidth))&bitMask),
-			byte((word>>(7*bitWidth))&bitMask),
-		)
-
-		i = j
-	}
-}
diff --git a/encoding/rle/rle_amd64.go b/encoding/rle/rle_amd64.go
index 960a2b3..37ed50b 100644
--- a/encoding/rle/rle_amd64.go
+++ b/encoding/rle/rle_amd64.go
@@ -9,8 +9,6 @@ import (
 var (
 	encodeInt32IndexEqual8Contiguous func(words [][8]int32) int
 	encodeInt32Bitpack               func(dst []byte, src [][8]int32, bitWidth uint) int
-	encodeBytesBitpack               func(dst []byte, src []uint64, bitWidth uint) int
-	decodeBytesBitpack               func(dst, src []byte, count, bitWidth uint)
 )
 
 func init() {
@@ -22,19 +20,10 @@ func init() {
 		encodeInt32IndexEqual8Contiguous = encodeInt32IndexEqual8ContiguousSSE
 		encodeInt32Bitpack = encodeInt32BitpackDefault
 	}
-
-	switch {
-	case cpu.X86.HasBMI2:
-		encodeBytesBitpack = encodeBytesBitpackBMI2
-		decodeBytesBitpack = decodeBytesBitpackBMI2
-	default:
-		encodeBytesBitpack = encodeBytesBitpackDefault
-		decodeBytesBitpack = decodeBytesBitpackDefault
-	}
 }
 
 //go:noescape
-func encodeBytesBitpackBMI2(dst []byte, src []uint64, bitWidth uint) int
+func encodeBytesBitpack(dst []byte, src []uint64, bitWidth uint) int
 
 //go:noescape
 func encodeInt32IndexEqual8ContiguousAVX2(words [][8]int32) int
@@ -57,4 +46,4 @@ func encodeInt32BitpackAVX2(dst []byte, src [][8]int32, bitWidth uint) int {
 }
 
 //go:noescape
-func decodeBytesBitpackBMI2(dst, src []byte, count, bitWidth uint)
+func decodeBytesBitpack(dst, src []byte, count, bitWidth uint)
diff --git a/encoding/rle/rle_amd64.s b/encoding/rle/rle_amd64.s
index a03a559..7e46e77 100644
--- a/encoding/rle/rle_amd64.s
+++ b/encoding/rle/rle_amd64.s
@@ -12,8 +12,8 @@ DATA bitMasks<>+40(SB)/8, $0b001111110011111100111111001111110011111100111111001
 DATA bitMasks<>+48(SB)/8, $0b0111111101111111011111110111111101111111011111110111111101111111
 DATA bitMasks<>+56(SB)/8, $0b1111111111111111111111111111111111111111111111111111111111111111
 
-// func decodeBytesBitpackBMI2(dst, src []byte, count, bitWidth uint)
-TEXT ·decodeBytesBitpackBMI2(SB), NOSPLIT, $0-64
+// func decodeBytesBitpack(dst, src []byte, count, bitWidth uint)
+TEXT ·decodeBytesBitpack(SB), NOSPLIT, $0-64
     MOVQ dst_base+0(FP), AX
     MOVQ src_base+24(FP), BX
     MOVQ count+48(FP), CX
@@ -34,8 +34,8 @@ test:
     JNE loop
     RET
 
-// func encodeBytesBitpackBMI2(dst []byte, src []uint64, bitWidth uint) int
-TEXT ·encodeBytesBitpackBMI2(SB), NOSPLIT, $0-64
+// func encodeBytesBitpack(dst []byte, src []uint64, bitWidth uint) int
+TEXT ·encodeBytesBitpack(SB), NOSPLIT, $0-64
     MOVQ dst_base+0(FP), AX
     MOVQ src_base+24(FP), BX
     MOVQ src_len+32(FP), CX
diff --git a/encoding/rle/rle_purego.go b/encoding/rle/rle_purego.go
index 8f3462d..f7584ea 100644
--- a/encoding/rle/rle_purego.go
+++ b/encoding/rle/rle_purego.go
@@ -2,8 +2,30 @@
 
 package rle
 
+import (
+	"encoding/binary"
+
+	"github.com/segmentio/parquet-go/internal/bitpack"
+)
+
 func encodeBytesBitpack(dst []byte, src []uint64, bitWidth uint) int {
-	return encodeBytesBitpackDefault(dst, src, bitWidth)
+	bitMask := uint64(1<<bitWidth) - 1
+	n := 0
+
+	for _, word := range src {
+		word = (word & bitMask) |
+			(((word >> 8) & bitMask) << (1 * bitWidth)) |
+			(((word >> 16) & bitMask) << (2 * bitWidth)) |
+			(((word >> 24) & bitMask) << (3 * bitWidth)) |
+			(((word >> 32) & bitMask) << (4 * bitWidth)) |
+			(((word >> 40) & bitMask) << (5 * bitWidth)) |
+			(((word >> 48) & bitMask) << (6 * bitWidth)) |
+			(((word >> 56) & bitMask) << (7 * bitWidth))
+		binary.LittleEndian.PutUint64(dst[n:], word)
+		n += int(bitWidth)
+	}
+
+	return n
 }
 
 func encodeInt32IndexEqual8Contiguous(words [][8]int32) (n int) {
@@ -18,5 +40,29 @@ func encodeInt32Bitpack(dst []byte, src [][8]int32, bitWidth uint) int {
 }
 
 func decodeBytesBitpack(dst, src []byte, count, bitWidth uint) {
-	decodeBytesBitpackDefault(dst, src, count, bitWidth)
+	dst = dst[:0]
+
+	bitMask := uint64(1<<bitWidth) - 1
+	byteCount := bitpack.ByteCount(8 * bitWidth)
+
+	for i := 0; count > 0; count -= 8 {
+		j := i + byteCount
+
+		bits := [8]byte{}
+		copy(bits[:], src[i:j])
+		word := binary.LittleEndian.Uint64(bits[:])
+
+		dst = append(dst,
+			byte((word>>(0*bitWidth))&bitMask),
+			byte((word>>(1*bitWidth))&bitMask),
+			byte((word>>(2*bitWidth))&bitMask),
+			byte((word>>(3*bitWidth))&bitMask),
+			byte((word>>(4*bitWidth))&bitMask),
+			byte((word>>(5*bitWidth))&bitMask),
+			byte((word>>(6*bitWidth))&bitMask),
+			byte((word>>(7*bitWidth))&bitMask),
+		)
+
+		i = j
+	}
 }
diff --git a/file.go b/file.go
index 05d6dca..0302b26 100644
--- a/file.go
+++ b/file.go
@@ -54,9 +54,6 @@ func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 		return nil, fmt.Errorf("invalid magic header of parquet file: %q", b[:4])
 	}
 
-	if cast, ok := f.reader.(interface{ SetMagicFooterSection(offset, length int64) }); ok {
-		cast.SetMagicFooterSection(size-8, 8)
-	}
 	if _, err := r.ReadAt(b[:8], size-8); err != nil {
 		return nil, fmt.Errorf("reading magic footer of parquet file: %w", err)
 	}
@@ -67,9 +64,6 @@ func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 	footerSize := int64(binary.LittleEndian.Uint32(b[:4]))
 	footerData := make([]byte, footerSize)
 
-	if cast, ok := f.reader.(interface{ SetFooterSection(offset, length int64) }); ok {
-		cast.SetFooterSection(size-(footerSize+8), footerSize)
-	}
 	if _, err := f.reader.ReadAt(footerData, size-(footerSize+8)); err != nil {
 		return nil, fmt.Errorf("reading footer of parquet file: %w", err)
 	}
@@ -123,12 +117,6 @@ func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 						return nil, err
 					}
 					offset, _ = s.Seek(0, io.SeekCurrent)
-					if cast, ok := r.(interface{ SetBloomFilterSection(offset, length int64) }); ok {
-						bloomFilterOffset := c.chunk.MetaData.BloomFilterOffset
-						bloomFilterLength := (offset - bloomFilterOffset) + int64(h.NumBytes)
-						cast.SetBloomFilterSection(bloomFilterOffset, bloomFilterLength)
-					}
-
 					c.bloomFilter = newBloomFilter(r, offset, &h)
 				}
 			}
@@ -204,9 +192,6 @@ func (f *File) ReadPageIndex() ([]format.ColumnIndex, []format.OffsetIndex, erro
 	if columnIndexOffset > 0 {
 		columnIndexData := indexBuffer[:columnIndexLength]
 
-		if cast, ok := f.reader.(interface{ SetColumnIndexSection(offset, length int64) }); ok {
-			cast.SetColumnIndexSection(columnIndexOffset, columnIndexLength)
-		}
 		if _, err := f.reader.ReadAt(columnIndexData, columnIndexOffset); err != nil {
 			return nil, nil, fmt.Errorf("reading %d bytes column index at offset %d: %w", columnIndexLength, columnIndexOffset, err)
 		}
@@ -228,9 +213,6 @@ func (f *File) ReadPageIndex() ([]format.ColumnIndex, []format.OffsetIndex, erro
 	if offsetIndexOffset > 0 {
 		offsetIndexData := indexBuffer[:offsetIndexLength]
 
-		if cast, ok := f.reader.(interface{ SetOffsetIndexSection(offset, length int64) }); ok {
-			cast.SetOffsetIndexSection(offsetIndexOffset, offsetIndexLength)
-		}
 		if _, err := f.reader.ReadAt(offsetIndexData, offsetIndexOffset); err != nil {
 			return nil, nil, fmt.Errorf("reading %d bytes offset index at offset %d: %w", offsetIndexLength, offsetIndexOffset, err)
 		}
diff --git a/hashprobe/aeshash/aeshash.go b/hashprobe/aeshash/aeshash.go
new file mode 100644
index 0000000..f2b72b6
--- /dev/null
+++ b/hashprobe/aeshash/aeshash.go
@@ -0,0 +1,21 @@
+// Package aeshash implements hashing functions derived from the Go runtime's
+// internal hashing based on the support of AES encryption in CPU instructions.
+//
+// On architecture where the CPU does not provide instructions for AES
+// encryption, the aeshash.Enabled function always returns false, and attempting
+// to call any other function will trigger a panic.
+package aeshash
+
+import "github.com/segmentio/parquet-go/sparse"
+
+func MultiHash32(hashes []uintptr, values []uint32, seed uintptr) {
+	MultiHashUint32Array(hashes, sparse.MakeUint32Array(values), seed)
+}
+
+func MultiHash64(hashes []uintptr, values []uint64, seed uintptr) {
+	MultiHashUint64Array(hashes, sparse.MakeUint64Array(values), seed)
+}
+
+func MultiHash128(hashes []uintptr, values [][16]byte, seed uintptr) {
+	MultiHashUint128Array(hashes, sparse.MakeUint128Array(values), seed)
+}
diff --git a/hashprobe/aeshash/aeshash_amd64.go b/hashprobe/aeshash/aeshash_amd64.go
new file mode 100644
index 0000000..de765fa
--- /dev/null
+++ b/hashprobe/aeshash/aeshash_amd64.go
@@ -0,0 +1,34 @@
+//go:build !purego
+
+package aeshash
+
+import (
+	"github.com/segmentio/parquet-go/sparse"
+	"golang.org/x/sys/cpu"
+)
+
+// Enabled returns true if AES hash is available on the system.
+//
+// The function uses the same logic than the Go runtime since we depend on
+// the AES hash state being initialized.
+//
+// See https://go.dev/src/runtime/alg.go
+func Enabled() bool { return cpu.X86.HasAES && cpu.X86.HasSSSE3 && cpu.X86.HasSSE41 }
+
+//go:noescape
+func Hash32(value uint32, seed uintptr) uintptr
+
+//go:noescape
+func Hash64(value uint64, seed uintptr) uintptr
+
+//go:noescape
+func Hash128(value [16]byte, seed uintptr) uintptr
+
+//go:noescape
+func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr)
+
+//go:noescape
+func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr)
+
+//go:noescape
+func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr)
diff --git a/hashprobe/aeshash/aeshash_amd64.s b/hashprobe/aeshash/aeshash_amd64.s
new file mode 100644
index 0000000..2145caa
--- /dev/null
+++ b/hashprobe/aeshash/aeshash_amd64.s
@@ -0,0 +1,155 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// func Hash32(value uint32, seed uintptr) uintptr
+TEXT ·Hash32(SB), NOSPLIT, $0-24
+    MOVL value+0(FP), AX
+    MOVQ seed+8(FP), BX
+
+    MOVOU runtime·aeskeysched+0(SB), X1
+    MOVOU runtime·aeskeysched+16(SB), X2
+    MOVOU runtime·aeskeysched+32(SB), X3
+
+    MOVQ BX, X0
+    PINSRD $2, AX, X0
+
+    AESENC X1, X0
+    AESENC X2, X0
+    AESENC X3, X0
+
+    MOVQ X0, ret+16(FP)
+    RET
+
+// func Hash64(value uint64, seed uintptr) uintptr
+TEXT ·Hash64(SB), NOSPLIT, $0-24
+    MOVQ value+0(FP), AX
+    MOVQ seed+8(FP), BX
+
+    MOVOU runtime·aeskeysched+0(SB), X1
+    MOVOU runtime·aeskeysched+16(SB), X2
+    MOVOU runtime·aeskeysched+32(SB), X3
+
+    MOVQ BX, X0
+    PINSRQ $1, AX, X0
+
+    AESENC X1, X0
+    AESENC X2, X0
+    AESENC X3, X0
+
+    MOVQ X0, ret+16(FP)
+    RET
+
+// func Hash128(value [16]byte, seed uintptr) uintptr
+TEXT ·Hash128(SB), NOSPLIT, $0-32
+    LEAQ value+0(FP), AX
+    MOVQ seed+16(FP), BX
+    MOVQ $16, CX
+
+    MOVQ BX, X0                      // 64 bits of per-table hash seed
+    PINSRW $4, CX, X0                // 16 bits of length
+    PSHUFHW $0, X0, X0               // repeat length 4 times total
+    PXOR runtime·aeskeysched(SB), X0 // xor in per-process seed
+    AESENC X0, X0                    // scramble seed
+
+    MOVOU (AX), X1
+    PXOR X0, X1
+    AESENC X1, X1
+    AESENC X1, X1
+    AESENC X1, X1
+
+    MOVQ X1, ret+24(FP)
+    RET
+
+// func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr)
+TEXT ·MultiHashUint32Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), AX
+    MOVQ values_array_ptr+24(FP), BX
+    MOVQ values_array_len+32(FP), CX
+    MOVQ values_array_off+40(FP), R8
+    MOVQ seed+48(FP), DX
+
+    MOVOU runtime·aeskeysched+0(SB), X1
+    MOVOU runtime·aeskeysched+16(SB), X2
+    MOVOU runtime·aeskeysched+32(SB), X3
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ DX, X0
+    PINSRD $2, (BX), X0
+
+    AESENC X1, X0
+    AESENC X2, X0
+    AESENC X3, X0
+
+    MOVQ X0, (AX)(SI*8)
+    INCQ SI
+    ADDQ R8, BX
+test:
+    CMPQ SI, CX
+    JNE loop
+    RET
+
+// func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr)
+TEXT ·MultiHashUint64Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), AX
+    MOVQ values_array_ptr+24(FP), BX
+    MOVQ values_array_len+32(FP), CX
+    MOVQ values_array_off+40(FP), R8
+    MOVQ seed+48(FP), DX
+
+    MOVOU runtime·aeskeysched+0(SB), X1
+    MOVOU runtime·aeskeysched+16(SB), X2
+    MOVOU runtime·aeskeysched+32(SB), X3
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ DX, X0
+    PINSRQ $1, (BX), X0
+
+    AESENC X1, X0
+    AESENC X2, X0
+    AESENC X3, X0
+
+    MOVQ X0, (AX)(SI*8)
+    INCQ SI
+    ADDQ R8, BX
+test:
+    CMPQ SI, CX
+    JNE loop
+    RET
+
+// func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr)
+TEXT ·MultiHashUint128Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), AX
+    MOVQ values_array_ptr+24(FP), BX
+    MOVQ values_array_len+32(FP), CX
+    MOVQ values_array_off+40(FP), R8
+    MOVQ seed+48(FP), DX
+    MOVQ $16, DI
+
+    MOVQ DX, X0
+    PINSRW $4, DI, X0
+    PSHUFHW $0, X0, X0
+    PXOR runtime·aeskeysched(SB), X0
+    AESENC X0, X0
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVOU (BX), X1
+
+    PXOR X0, X1
+    AESENC X1, X1
+    AESENC X1, X1
+    AESENC X1, X1
+
+    MOVQ X1, (AX)(SI*8)
+    INCQ SI
+    ADDQ R8, BX
+test:
+    CMPQ SI, CX
+    JNE loop
+    RET
diff --git a/hashprobe/aeshash/aeshash_purego.go b/hashprobe/aeshash/aeshash_purego.go
new file mode 100644
index 0000000..c5f5774
--- /dev/null
+++ b/hashprobe/aeshash/aeshash_purego.go
@@ -0,0 +1,29 @@
+//go:build purego || !amd64
+
+package aeshash
+
+import "github.com/segmentio/parquet-go/sparse"
+
+// Enabled always returns false since we assume that AES instructions are not
+// available by default.
+func Enabled() bool { return false }
+
+const unsupported = "BUG: AES hash is not available on this platform"
+
+func Hash32(value uint32, seed uintptr) uintptr { panic(unsupported) }
+
+func Hash64(value uint64, seed uintptr) uintptr { panic(unsupported) }
+
+func Hash128(value [16]byte, seed uintptr) uintptr { panic(unsupported) }
+
+func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr) {
+	panic(unsupported)
+}
+
+func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr) {
+	panic(unsupported)
+}
+
+func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr) {
+	panic(unsupported)
+}
diff --git a/hashprobe/hashprobe.go b/hashprobe/hashprobe.go
new file mode 100644
index 0000000..9e61aab
--- /dev/null
+++ b/hashprobe/hashprobe.go
@@ -0,0 +1,779 @@
+// Package hashprobe provides implementations of probing tables for various
+// data types.
+//
+// Probing tables are specialized hash tables supporting only a single
+// "probing" operation which behave like a "lookup or insert". When a key
+// is probed, either its value is retrieved if it already existed in the table,
+// or it is inserted and assigned its index in the insert sequence as value.
+//
+// Values are represented as signed 32 bits integers, which means that probing
+// tables defined in this package may contain at most 2^31-1 entries.
+//
+// Probing tables have a method named Probe with the following signature:
+//
+//	func (t *Int64Table) Probe(keys []int64, values []int32) int {
+//		...
+//	}
+//
+// The method takes an array of keys to probe as first argument, an array of
+// values where the indexes of each key will be written as second argument, and
+// returns the number of keys that were inserted during the call.
+//
+// Applications that need to determine which keys were inserted can capture the
+// length of the probing table prior to the call, and scan the list of values
+// looking for indexes greater or equal to the length of the table before the
+// call.
+package hashprobe
+
+import (
+	cryptoRand "crypto/rand"
+	"encoding/binary"
+	"math"
+	"math/bits"
+	"math/rand"
+	"sync"
+
+	"github.com/segmentio/parquet-go/hashprobe/aeshash"
+	"github.com/segmentio/parquet-go/hashprobe/wyhash"
+	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
+)
+
+const (
+	// Number of probes tested per iteration. This parameter balances between
+	// the amount of memory allocated on the stack to hold the computed hashes
+	// of the keys being probed, and amortizing the baseline cost of the probing
+	// algorithm.
+	//
+	// The larger the value, the more memory is required, but lower the baseline
+	// cost will be.
+	//
+	// We chose a value that is somewhat large, resulting in reserving 2KiB of
+	// stack but mostly erasing the baseline cost.
+	probesPerLoop = 256
+)
+
+var (
+	prngSeed   [8]byte
+	prngMutex  sync.Mutex
+	prngSource rand.Source64
+)
+
+func init() {
+	_, err := cryptoRand.Read(prngSeed[:])
+	if err != nil {
+		panic("cannot seed random number generator from system source: " + err.Error())
+	}
+	seed := int64(binary.LittleEndian.Uint64(prngSeed[:]))
+	prngSource = rand.NewSource(seed).(rand.Source64)
+}
+
+func nextPowerOf2(n int) int {
+	return 1 << (64 - bits.LeadingZeros64(uint64(n-1)))
+}
+
+func randSeed() uintptr {
+	prngMutex.Lock()
+	defer prngMutex.Unlock()
+	return uintptr(prngSource.Uint64())
+}
+
+type Int32Table struct{ table32 }
+
+func NewInt32Table(cap int, maxLoad float64) *Int32Table {
+	return &Int32Table{makeTable32(cap, maxLoad)}
+}
+
+func (t *Int32Table) Reset() { t.reset() }
+
+func (t *Int32Table) Len() int { return t.len }
+
+func (t *Int32Table) Cap() int { return t.size() }
+
+func (t *Int32Table) Probe(keys, values []int32) int {
+	return t.probe(unsafecast.Int32ToUint32(keys), values)
+}
+
+func (t *Int32Table) ProbeArray(keys sparse.Int32Array, values []int32) int {
+	return t.probeArray(keys.Uint32Array(), values)
+}
+
+type Float32Table struct{ table32 }
+
+func NewFloat32Table(cap int, maxLoad float64) *Float32Table {
+	return &Float32Table{makeTable32(cap, maxLoad)}
+}
+
+func (t *Float32Table) Reset() { t.reset() }
+
+func (t *Float32Table) Len() int { return t.len }
+
+func (t *Float32Table) Cap() int { return t.size() }
+
+func (t *Float32Table) Probe(keys []float32, values []int32) int {
+	return t.probe(unsafecast.Float32ToUint32(keys), values)
+}
+
+func (t *Float32Table) ProbeArray(keys sparse.Float32Array, values []int32) int {
+	return t.probeArray(keys.Uint32Array(), values)
+}
+
+type Uint32Table struct{ table32 }
+
+func NewUint32Table(cap int, maxLoad float64) *Uint32Table {
+	return &Uint32Table{makeTable32(cap, maxLoad)}
+}
+
+func (t *Uint32Table) Reset() { t.reset() }
+
+func (t *Uint32Table) Len() int { return t.len }
+
+func (t *Uint32Table) Cap() int { return t.size() }
+
+func (t *Uint32Table) Probe(keys []uint32, values []int32) int {
+	return t.probe(keys, values)
+}
+
+func (t *Uint32Table) ProbeArray(keys sparse.Uint32Array, values []int32) int {
+	return t.probeArray(keys, values)
+}
+
+// table32 is the generic implementation of probing tables for 32 bit types.
+//
+// The table uses the following memory layout:
+//
+//		[group 0][group 1][...][group N]
+//
+// Each group contains up to 7 key/value pairs, and is exactly 64 bytes in size,
+// which allows it to fit within a single cache line, and ensures that probes
+// can be performed with a single memory load per key.
+//
+// Groups fill up by appending new entries to the keys and values arrays. When a
+// group is full, the probe checks the next group.
+//
+// https://en.wikipedia.org/wiki/Linear_probing
+type table32 struct {
+	len     int
+	maxLen  int
+	maxLoad float64
+	seed    uintptr
+	table   []table32Group
+}
+
+const table32GroupSize = 7
+
+type table32Group struct {
+	keys   [table32GroupSize]uint32
+	values [table32GroupSize]uint32
+	bits   uint32
+	_      uint32
+}
+
+func makeTable32(cap int, maxLoad float64) (t table32) {
+	if maxLoad < 0 || maxLoad > 1 {
+		panic("max load of probing table must be a value between 0 and 1")
+	}
+	if cap < table32GroupSize {
+		cap = table32GroupSize
+	}
+	t.init(cap, maxLoad)
+	return t
+}
+
+func (t *table32) size() int {
+	return table32GroupSize * len(t.table)
+}
+
+func (t *table32) init(cap int, maxLoad float64) {
+	m := int(math.Ceil((1 / maxLoad) * float64(cap)))
+	n := nextPowerOf2((m + (table32GroupSize - 1)) / table32GroupSize)
+	*t = table32{
+		maxLen:  int(math.Ceil(maxLoad * float64(table32GroupSize*n))),
+		maxLoad: maxLoad,
+		seed:    randSeed(),
+		table:   make([]table32Group, n),
+	}
+}
+
+func (t *table32) grow(totalValues int) {
+	tmp := table32{}
+	tmp.init(totalValues, t.maxLoad)
+	tmp.len = t.len
+
+	hashes := make([]uintptr, table32GroupSize)
+	modulo := uintptr(len(tmp.table)) - 1
+
+	for i := range t.table {
+		g := &t.table[i]
+		n := bits.OnesCount32(g.bits)
+
+		if aeshash.Enabled() {
+			aeshash.MultiHash32(hashes[:n], g.keys[:n], tmp.seed)
+		} else {
+			wyhash.MultiHash32(hashes[:n], g.keys[:n], tmp.seed)
+		}
+
+		for j, hash := range hashes[:n] {
+			for {
+				group := &tmp.table[hash&modulo]
+
+				if n := bits.OnesCount32(group.bits); n < table32GroupSize {
+					group.bits = (group.bits << 1) | 1
+					group.keys[n] = g.keys[j]
+					group.values[n] = g.values[j]
+					break
+				}
+
+				hash++
+			}
+		}
+	}
+
+	*t = tmp
+}
+
+func (t *table32) reset() {
+	t.len = 0
+
+	for i := range t.table {
+		t.table[i] = table32Group{}
+	}
+}
+
+func (t *table32) probe(keys []uint32, values []int32) int {
+	return t.probeArray(sparse.MakeUint32Array(keys), values)
+}
+
+func (t *table32) probeArray(keys sparse.Uint32Array, values []int32) int {
+	numKeys := keys.Len()
+
+	if totalValues := t.len + numKeys; totalValues > t.maxLen {
+		t.grow(totalValues)
+	}
+
+	var hashes [probesPerLoop]uintptr
+	var baseLength = t.len
+	var useAesHash = aeshash.Enabled()
+
+	_ = values[:numKeys]
+
+	for i := 0; i < numKeys; {
+		j := len(hashes) + i
+		n := len(hashes)
+
+		if j > numKeys {
+			j = numKeys
+			n = numKeys - i
+		}
+
+		k := keys.Slice(i, j)
+		v := values[i:j:j]
+		h := hashes[:n:n]
+
+		if useAesHash {
+			aeshash.MultiHashUint32Array(h, k, t.seed)
+		} else {
+			wyhash.MultiHashUint32Array(h, k, t.seed)
+		}
+
+		t.len = multiProbe32(t.table, t.len, h, k, v)
+		i = j
+	}
+
+	return t.len - baseLength
+}
+
+func multiProbe32Default(table []table32Group, numKeys int, hashes []uintptr, keys sparse.Uint32Array, values []int32) int {
+	modulo := uintptr(len(table)) - 1
+
+	for i, hash := range hashes {
+		key := keys.Index(i)
+		for {
+			group := &table[hash&modulo]
+			index := table32GroupSize
+			value := int32(0)
+
+			for j, k := range group.keys {
+				if k == key {
+					index = j
+					break
+				}
+			}
+
+			if n := bits.OnesCount32(group.bits); index < n {
+				value = int32(group.values[index])
+			} else {
+				if n == table32GroupSize {
+					hash++
+					continue
+				}
+
+				value = int32(numKeys)
+				group.bits = (group.bits << 1) | 1
+				group.keys[n] = key
+				group.values[n] = uint32(value)
+				numKeys++
+			}
+
+			values[i] = value
+			break
+		}
+	}
+
+	return numKeys
+}
+
+type Int64Table struct{ table64 }
+
+func NewInt64Table(cap int, maxLoad float64) *Int64Table {
+	return &Int64Table{makeTable64(cap, maxLoad)}
+}
+
+func (t *Int64Table) Reset() { t.reset() }
+
+func (t *Int64Table) Len() int { return t.len }
+
+func (t *Int64Table) Cap() int { return t.size() }
+
+func (t *Int64Table) Probe(keys []int64, values []int32) int {
+	return t.probe(unsafecast.Int64ToUint64(keys), values)
+}
+
+func (t *Int64Table) ProbeArray(keys sparse.Int64Array, values []int32) int {
+	return t.probeArray(keys.Uint64Array(), values)
+}
+
+type Float64Table struct{ table64 }
+
+func NewFloat64Table(cap int, maxLoad float64) *Float64Table {
+	return &Float64Table{makeTable64(cap, maxLoad)}
+}
+
+func (t *Float64Table) Reset() { t.reset() }
+
+func (t *Float64Table) Len() int { return t.len }
+
+func (t *Float64Table) Cap() int { return t.size() }
+
+func (t *Float64Table) Probe(keys []float64, values []int32) int {
+	return t.probe(unsafecast.Float64ToUint64(keys), values)
+}
+
+func (t *Float64Table) ProbeArray(keys sparse.Float64Array, values []int32) int {
+	return t.probeArray(keys.Uint64Array(), values)
+}
+
+type Uint64Table struct{ table64 }
+
+func NewUint64Table(cap int, maxLoad float64) *Uint64Table {
+	return &Uint64Table{makeTable64(cap, maxLoad)}
+}
+
+func (t *Uint64Table) Reset() { t.reset() }
+
+func (t *Uint64Table) Len() int { return t.len }
+
+func (t *Uint64Table) Cap() int { return t.size() }
+
+func (t *Uint64Table) Probe(keys []uint64, values []int32) int {
+	return t.probe(keys, values)
+}
+
+func (t *Uint64Table) ProbeArray(keys sparse.Uint64Array, values []int32) int {
+	return t.probeArray(keys, values)
+}
+
+// table64 is the generic implementation of probing tables for 64 bit types.
+//
+// The table uses a layout similar to the one documented on the table for 32 bit
+// keys (see table32). Each group holds up to 4 key/value pairs (instead of 7
+// like table32) so that each group fits in a single CPU cache line. This table
+// version has a bit lower memory density, with ~23% of table memory being used
+// for padding.
+//
+// Technically we could hold up to 5 entries per group and still fit within the
+// 64 bytes of a CPU cache line; on x86 platforms, AVX2 registers can only hold
+// four 64 bit values, we would need twice as many instructions per probe if the
+// groups were holding 5 values. The trade off of memory for compute efficiency
+// appeared to be the right choice at the time.
+type table64 struct {
+	len     int
+	maxLen  int
+	maxLoad float64
+	seed    uintptr
+	table   []table64Group
+}
+
+const table64GroupSize = 4
+
+type table64Group struct {
+	keys   [table64GroupSize]uint64
+	values [table64GroupSize]uint32
+	bits   uint32
+	_      uint32
+	_      uint32
+	_      uint32
+}
+
+func makeTable64(cap int, maxLoad float64) (t table64) {
+	if maxLoad < 0 || maxLoad > 1 {
+		panic("max load of probing table must be a value between 0 and 1")
+	}
+	if cap < table64GroupSize {
+		cap = table64GroupSize
+	}
+	t.init(cap, maxLoad)
+	return t
+}
+
+func (t *table64) size() int {
+	return table64GroupSize * len(t.table)
+}
+
+func (t *table64) init(cap int, maxLoad float64) {
+	m := int(math.Ceil((1 / maxLoad) * float64(cap)))
+	n := nextPowerOf2((m + (table64GroupSize - 1)) / table64GroupSize)
+	*t = table64{
+		maxLen:  int(math.Ceil(maxLoad * float64(table64GroupSize*n))),
+		maxLoad: maxLoad,
+		seed:    randSeed(),
+		table:   make([]table64Group, n),
+	}
+}
+
+func (t *table64) grow(totalValues int) {
+	tmp := table64{}
+	tmp.init(totalValues, t.maxLoad)
+	tmp.len = t.len
+
+	hashes := make([]uintptr, table64GroupSize)
+	modulo := uintptr(len(tmp.table)) - 1
+
+	for i := range t.table {
+		g := &t.table[i]
+		n := bits.OnesCount32(g.bits)
+
+		if aeshash.Enabled() {
+			aeshash.MultiHash64(hashes[:n], g.keys[:n], tmp.seed)
+		} else {
+			wyhash.MultiHash64(hashes[:n], g.keys[:n], tmp.seed)
+		}
+
+		for j, hash := range hashes[:n] {
+			for {
+				group := &tmp.table[hash&modulo]
+
+				if n := bits.OnesCount32(group.bits); n < table64GroupSize {
+					group.bits = (group.bits << 1) | 1
+					group.keys[n] = g.keys[j]
+					group.values[n] = g.values[j]
+					break
+				}
+
+				hash++
+			}
+		}
+	}
+
+	*t = tmp
+}
+
+func (t *table64) reset() {
+	t.len = 0
+
+	for i := range t.table {
+		t.table[i] = table64Group{}
+	}
+}
+
+func (t *table64) probe(keys []uint64, values []int32) int {
+	return t.probeArray(sparse.MakeUint64Array(keys), values)
+}
+
+func (t *table64) probeArray(keys sparse.Uint64Array, values []int32) int {
+	numKeys := keys.Len()
+
+	if totalValues := t.len + numKeys; totalValues > t.maxLen {
+		t.grow(totalValues)
+	}
+
+	var hashes [probesPerLoop]uintptr
+	var baseLength = t.len
+	var useAesHash = aeshash.Enabled()
+
+	_ = values[:numKeys]
+
+	for i := 0; i < numKeys; {
+		j := len(hashes) + i
+		n := len(hashes)
+
+		if j > numKeys {
+			j = numKeys
+			n = numKeys - i
+		}
+
+		k := keys.Slice(i, j)
+		v := values[i:j:j]
+		h := hashes[:n:n]
+
+		if useAesHash {
+			aeshash.MultiHashUint64Array(h, k, t.seed)
+		} else {
+			wyhash.MultiHashUint64Array(h, k, t.seed)
+		}
+
+		t.len = multiProbe64(t.table, t.len, h, k, v)
+		i = j
+	}
+
+	return t.len - baseLength
+}
+
+func multiProbe64Default(table []table64Group, numKeys int, hashes []uintptr, keys sparse.Uint64Array, values []int32) int {
+	modulo := uintptr(len(table)) - 1
+
+	for i, hash := range hashes {
+		key := keys.Index(i)
+		for {
+			group := &table[hash&modulo]
+			index := table64GroupSize
+			value := int32(0)
+
+			for i, k := range group.keys {
+				if k == key {
+					index = i
+					break
+				}
+			}
+
+			if n := bits.OnesCount32(group.bits); index < n {
+				value = int32(group.values[index])
+			} else {
+				if n == table64GroupSize {
+					hash++
+					continue
+				}
+
+				value = int32(numKeys)
+				group.bits = (group.bits << 1) | 1
+				group.keys[n] = key
+				group.values[n] = uint32(value)
+				numKeys++
+			}
+
+			values[i] = value
+			break
+		}
+	}
+
+	return numKeys
+}
+
+type Uint128Table struct{ table128 }
+
+func NewUint128Table(cap int, maxLoad float64) *Uint128Table {
+	return &Uint128Table{makeTable128(cap, maxLoad)}
+}
+
+func (t *Uint128Table) Reset() { t.reset() }
+
+func (t *Uint128Table) Len() int { return t.len }
+
+func (t *Uint128Table) Cap() int { return t.cap }
+
+func (t *Uint128Table) Probe(keys [][16]byte, values []int32) int {
+	return t.probe(keys, values)
+}
+
+func (t *Uint128Table) ProbeArray(keys sparse.Uint128Array, values []int32) int {
+	return t.probeArray(keys, values)
+}
+
+// table128 is the generic implementation of probing tables for 128 bit types.
+//
+// This table uses the following memory layout:
+//
+//		[key A][key B][...][value A][value B][...]
+//
+// The table stores values as their actual value plus one, and uses zero as a
+// sentinel to determine whether a slot is occupied. A linear probing strategy
+// is used to resolve conflicts. This approach results in at most two memory
+// loads for every four keys being tested, since the location of a key and its
+// corresponding value will not be contiguous on the same CPU cache line, but
+// a cache line can hold four 16 byte keys.
+type table128 struct {
+	len     int
+	cap     int
+	maxLen  int
+	maxLoad float64
+	seed    uintptr
+	table   []byte
+}
+
+func makeTable128(cap int, maxLoad float64) (t table128) {
+	if maxLoad < 0 || maxLoad > 1 {
+		panic("max load of probing table must be a value between 0 and 1")
+	}
+	if cap < 8 {
+		cap = 8
+	}
+	t.init(cap, maxLoad)
+	return t
+}
+
+func (t *table128) init(cap int, maxLoad float64) {
+	m := int(math.Ceil((1 / maxLoad) * float64(cap)))
+	n := nextPowerOf2(m)
+	*t = table128{
+		cap:     n,
+		maxLen:  int(math.Ceil(maxLoad * float64(n))),
+		maxLoad: maxLoad,
+		seed:    randSeed(),
+		table:   make([]byte, 16*n+4*n),
+	}
+}
+
+func (t *table128) kv() (keys [][16]byte, values []int32) {
+	i := t.cap * 16
+	return unsafecast.BytesToUint128(t.table[:i]), unsafecast.BytesToInt32(t.table[i:])
+}
+
+func (t *table128) grow(totalValues int) {
+	tmp := table128{}
+	tmp.init(totalValues, t.maxLoad)
+	tmp.len = t.len
+
+	keys, values := t.kv()
+	hashes := make([]uintptr, probesPerLoop)
+	useAesHash := aeshash.Enabled()
+
+	_ = values[:len(keys)]
+
+	for i := 0; i < len(keys); {
+		j := len(hashes) + i
+		n := len(hashes)
+
+		if j > len(keys) {
+			j = len(keys)
+			n = len(keys) - i
+		}
+
+		h := hashes[:n:n]
+		k := keys[i:j:j]
+		v := values[i:j:j]
+
+		if useAesHash {
+			aeshash.MultiHash128(h, k, tmp.seed)
+		} else {
+			wyhash.MultiHash128(h, k, tmp.seed)
+		}
+
+		tmp.insert(h, k, v)
+		i = j
+	}
+
+	*t = tmp
+}
+
+func (t *table128) insert(hashes []uintptr, keys [][16]byte, values []int32) {
+	tableKeys, tableValues := t.kv()
+	modulo := uintptr(t.cap) - 1
+
+	for i, hash := range hashes {
+		for {
+			j := hash & modulo
+			v := tableValues[j]
+
+			if v == 0 {
+				tableKeys[j] = keys[i]
+				tableValues[j] = values[i]
+				break
+			}
+
+			hash++
+		}
+	}
+}
+
+func (t *table128) reset() {
+	t.len = 0
+
+	for i := range t.table {
+		t.table[i] = 0
+	}
+}
+
+func (t *table128) probe(keys [][16]byte, values []int32) int {
+	return t.probeArray(sparse.MakeUint128Array(keys), values)
+}
+
+func (t *table128) probeArray(keys sparse.Uint128Array, values []int32) int {
+	numKeys := keys.Len()
+
+	if totalValues := t.len + numKeys; totalValues > t.maxLen {
+		t.grow(totalValues)
+	}
+
+	var hashes [probesPerLoop]uintptr
+	var baseLength = t.len
+	var useAesHash = aeshash.Enabled()
+
+	_ = values[:numKeys]
+
+	for i := 0; i < numKeys; {
+		j := len(hashes) + i
+		n := len(hashes)
+
+		if j > numKeys {
+			j = numKeys
+			n = numKeys - i
+		}
+
+		k := keys.Slice(i, j)
+		v := values[i:j:j]
+		h := hashes[:n:n]
+
+		if useAesHash {
+			aeshash.MultiHashUint128Array(h, k, t.seed)
+		} else {
+			wyhash.MultiHashUint128Array(h, k, t.seed)
+		}
+
+		t.len = multiProbe128(t.table, t.cap, t.len, h, k, v)
+		i = j
+	}
+
+	return t.len - baseLength
+}
+
+func multiProbe128Default(table []byte, tableCap, tableLen int, hashes []uintptr, keys sparse.Uint128Array, values []int32) int {
+	modulo := uintptr(tableCap) - 1
+	offset := uintptr(tableCap) * 16
+	tableKeys := unsafecast.BytesToUint128(table[:offset])
+	tableValues := unsafecast.BytesToInt32(table[offset:])
+
+	for i, hash := range hashes {
+		key := keys.Index(i)
+		for {
+			j := hash & modulo
+			v := tableValues[j]
+
+			if v == 0 {
+				values[i] = int32(tableLen)
+				tableLen++
+				tableKeys[j] = key
+				tableValues[j] = int32(tableLen)
+				break
+			}
+
+			if key == tableKeys[j] {
+				values[i] = v - 1
+				break
+			}
+
+			hash++
+		}
+	}
+
+	return tableLen
+}
diff --git a/hashprobe/hashprobe_amd64.go b/hashprobe/hashprobe_amd64.go
new file mode 100644
index 0000000..451acd3
--- /dev/null
+++ b/hashprobe/hashprobe_amd64.go
@@ -0,0 +1,38 @@
+//go:build !purego
+
+package hashprobe
+
+import (
+	"github.com/segmentio/parquet-go/sparse"
+	"golang.org/x/sys/cpu"
+)
+
+//go:noescape
+func multiProbe32AVX2(table []table32Group, numKeys int, hashes []uintptr, keys sparse.Uint32Array, values []int32) int
+
+//go:noescape
+func multiProbe64AVX2(table []table64Group, numKeys int, hashes []uintptr, keys sparse.Uint64Array, values []int32) int
+
+//go:noescape
+func multiProbe128SSE2(table []byte, tableCap, tableLen int, hashes []uintptr, keys sparse.Uint128Array, values []int32) int
+
+func multiProbe32(table []table32Group, numKeys int, hashes []uintptr, keys sparse.Uint32Array, values []int32) int {
+	if cpu.X86.HasAVX2 {
+		return multiProbe32AVX2(table, numKeys, hashes, keys, values)
+	}
+	return multiProbe32Default(table, numKeys, hashes, keys, values)
+}
+
+func multiProbe64(table []table64Group, numKeys int, hashes []uintptr, keys sparse.Uint64Array, values []int32) int {
+	if cpu.X86.HasAVX2 {
+		return multiProbe64AVX2(table, numKeys, hashes, keys, values)
+	}
+	return multiProbe64Default(table, numKeys, hashes, keys, values)
+}
+
+func multiProbe128(table []byte, tableCap, tableLen int, hashes []uintptr, keys sparse.Uint128Array, values []int32) int {
+	if cpu.X86.HasSSE2 {
+		return multiProbe128SSE2(table, tableCap, tableLen, hashes, keys, values)
+	}
+	return multiProbe128Default(table, tableCap, tableLen, hashes, keys, values)
+}
diff --git a/hashprobe/hashprobe_amd64.s b/hashprobe/hashprobe_amd64.s
new file mode 100644
index 0000000..5c5c0c0
--- /dev/null
+++ b/hashprobe/hashprobe_amd64.s
@@ -0,0 +1,197 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// This version of the probing algorithm for 32 bit keys takes advantage of
+// the memory layout of table groups and SIMD instructions to accelerate the
+// probing operations.
+//
+// The first 32 bytes of a table group contain the bit mask indicating which
+// slots are in use, and the array of keys, which fits into a single vector
+// register (YMM) and can be loaded and tested with a single instruction.
+//
+// A first version of the table group used the number of keys held in the
+// group instead of a bit mask, which required the probing operation to
+// reconstruct the bit mask during the lookup operation in order to identify
+// which elements of the VPCMPEQD result should be retained. The extra CPU
+// instructions used to reconstruct the bit mask had a measurable overhead.
+// By holding the bit mask in the data structure, we can determine the number
+// of keys in a group using the POPCNT instruction, and avoid recomputing the
+// mask during lookups.
+//
+// func multiProbe32AVX2(table []table32Group, numKeys int, hashes []uintptr, keys sparse.Uint32Array, values []int32) int
+TEXT ·multiProbe32AVX2(SB), NOSPLIT, $0-112
+    MOVQ table_base+0(FP), AX
+    MOVQ table_len+8(FP), BX
+    MOVQ numKeys+24(FP), CX
+    MOVQ hashes_base+32(FP), DX
+    MOVQ hashes_len+40(FP), DI
+    MOVQ keys_array_ptr+56(FP), R8
+    MOVQ keys_array_off+72(FP), R15
+    MOVQ values_base+80(FP), R9
+    DECQ BX // modulo = len(table) - 1
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ (DX)(SI*8), R10  // hash
+    VPBROADCASTD (R8), Y0 // [key]
+probe:
+    MOVQ R10, R11
+    ANDQ BX, R11 // hash & modulo
+    SHLQ $6, R11 // x 64 (size of table32Group)
+    LEAQ (AX)(R11*1), R12
+
+    VMOVDQU (R12), Y1
+    VPCMPEQD Y0, Y1, Y2
+    VMOVMSKPS Y2, R11
+    MOVL 56(R12), R13
+    TESTL R11, R13
+    JZ insert
+
+    TZCNTL R11, R13
+    MOVL 28(R12)(R13*4), R14
+next:
+    MOVL R14, (R9)(SI*4)
+    INCQ SI
+    ADDQ R15, R8
+test:
+    CMPQ SI, DI
+    JNE loop
+    MOVQ CX, ret+104(FP)
+    VZEROUPPER
+    RET
+insert:
+    CMPL R13, $0b1111111
+    JE probeNextGroup
+
+    MOVL R13, R11
+    POPCNTL R13, R13
+    MOVQ X0, R14 // key
+    SHLL $1, R11
+    ORL $1, R11
+    MOVL R11, 56(R12)       // group.len = (group.len << 1) | 1
+    MOVL R14, (R12)(R13*4)  // group.keys[i] = key
+    MOVL CX, 28(R12)(R13*4) // group.values[i] = value
+    MOVL CX, R14
+    INCL CX
+    JMP next
+probeNextGroup:
+    INCQ R10
+    JMP probe
+
+// func multiProbe64AVX2(table []table64Group, numKeys int, hashes []uintptr, keys sparse.Uint64Array, values []int32) int
+TEXT ·multiProbe64AVX2(SB), NOSPLIT, $0-112
+    MOVQ table_base+0(FP), AX
+    MOVQ table_len+8(FP), BX
+    MOVQ numKeys+24(FP), CX
+    MOVQ hashes_base+32(FP), DX
+    MOVQ hashes_len+40(FP), DI
+    MOVQ keys_array_ptr+56(FP), R8
+    MOVQ keys_array_off+72(FP), R15
+    MOVQ values_base+80(FP), R9
+    DECQ BX // modulo = len(table) - 1
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ (DX)(SI*8), R10        // hash
+    VPBROADCASTQ (R8), Y0 // [key]
+probe:
+    MOVQ R10, R11
+    ANDQ BX, R11 // hash & modulo
+    SHLQ $6, R11 // x 64 (size of table64Group)
+    LEAQ (AX)(R11*1), R12
+
+    VMOVDQU (R12), Y1
+    VPCMPEQQ Y0, Y1, Y2
+    VMOVMSKPD Y2, R11
+    MOVL 48(R12), R13
+    TESTL R11, R13
+    JZ insert
+
+    TZCNTL R11, R13
+    MOVL 32(R12)(R13*4), R14
+next:
+    MOVL R14, (R9)(SI*4)
+    INCQ SI
+    ADDQ R15, R8
+test:
+    CMPQ SI, DI
+    JNE loop
+    MOVQ CX, ret+104(FP)
+    VZEROUPPER
+    RET
+insert:
+    CMPL R13, $0b1111
+    JE probeNextGroup
+
+    MOVL R13, R11
+    POPCNTL R13, R13
+    SHLL $1, R11
+    ORL $1, R11
+    MOVL R11, 48(R12)       // group.len = (group.len << 1) | 1
+    MOVQ X0, (R12)(R13*8)   // group.keys[i] = key
+    MOVL CX, 32(R12)(R13*4) // group.values[i] = value
+    MOVL CX, R14
+    INCL CX
+    JMP next
+probeNextGroup:
+    INCQ R10
+    JMP probe
+
+// func multiProbe128SSE2(table []byte, tableCap, tableLen int, hashes []uintptr, keys sparse.Uint128Array, values []int32) int
+TEXT ·multiProbe128SSE2(SB), NOSPLIT, $0-120
+    MOVQ table_base+0(FP), AX
+    MOVQ tableCap+24(FP), BX
+    MOVQ tableLen+32(FP), CX
+    MOVQ hashes_base+40(FP), DX
+    MOVQ hashes_len+48(FP), DI
+    MOVQ keys_array_ptr+64(FP), R8
+    MOVQ keys_array_off+80(FP), R15
+    MOVQ values_base+88(FP), R9
+
+    MOVQ BX, R10
+    SHLQ $4, R10
+    LEAQ (AX)(R10*1), R10
+    DECQ BX // modulo = tableCap - 1
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ (DX)(SI*8), R11 // hash
+    MOVOU (R8), X0       // key
+probe:
+    MOVQ R11, R12
+    ANDQ BX, R12
+
+    MOVL (R10)(R12*4), R14
+    CMPL R14, $0
+    JE insert
+
+    SHLQ $4, R12
+    MOVOU (AX)(R12*1), X1
+    PCMPEQL X0, X1
+    MOVMSKPS X1, R13
+    CMPL R13, $0b1111
+    JE next
+
+    INCQ R11
+    JMP probe
+next:
+    DECL R14
+    MOVL R14, (R9)(SI*4)
+    INCQ SI
+    ADDQ R15, R8
+test:
+    CMPQ SI, DI
+    JNE loop
+    MOVQ CX, ret+112(FP)
+    RET
+insert:
+    INCL CX
+    MOVL CX, (R10)(R12*4)
+    MOVL CX, R14
+    SHLQ $4, R12
+    MOVOU X0, (AX)(R12*1)
+    JMP next
diff --git a/hashprobe/hashprobe_purego.go b/hashprobe/hashprobe_purego.go
new file mode 100644
index 0000000..1c81781
--- /dev/null
+++ b/hashprobe/hashprobe_purego.go
@@ -0,0 +1,17 @@
+//go:build purego || !amd64
+
+package hashprobe
+
+import "github.com/segmentio/parquet-go/sparse"
+
+func multiProbe32(table []table32Group, numKeys int, hashes []uintptr, keys sparse.Uint32Array, values []int32) int {
+	return multiProbe32Default(table, numKeys, hashes, keys, values)
+}
+
+func multiProbe64(table []table64Group, numKeys int, hashes []uintptr, keys sparse.Uint64Array, values []int32) int {
+	return multiProbe64Default(table, numKeys, hashes, keys, values)
+}
+
+func multiProbe128(table []byte, tableCap, tableLen int, hashes []uintptr, keys sparse.Uint128Array, values []int32) int {
+	return multiProbe128Default(table, tableCap, tableLen, hashes, keys, values)
+}
diff --git a/hashprobe/wyhash/wyhash.go b/hashprobe/wyhash/wyhash.go
new file mode 100644
index 0000000..a0ddbf0
--- /dev/null
+++ b/hashprobe/wyhash/wyhash.go
@@ -0,0 +1,49 @@
+// Package wyhash implements a hashing algorithm derived from the Go runtime's
+// internal hashing fallback, which uses a variation of the wyhash algorithm.
+package wyhash
+
+import (
+	"encoding/binary"
+	"math/bits"
+
+	"github.com/segmentio/parquet-go/sparse"
+)
+
+const (
+	m1 = 0xa0761d6478bd642f
+	m2 = 0xe7037ed1a0b428db
+	m3 = 0x8ebc6af09c88c6e3
+	m4 = 0x589965cc75374cc3
+	m5 = 0x1d8e4e27c47d124f
+)
+
+func mix(a, b uint64) uint64 {
+	hi, lo := bits.Mul64(a, b)
+	return hi ^ lo
+}
+
+func Hash32(value uint32, seed uintptr) uintptr {
+	return uintptr(mix(m5^4, mix(uint64(value)^m2, uint64(value)^uint64(seed)^m1)))
+}
+
+func Hash64(value uint64, seed uintptr) uintptr {
+	return uintptr(mix(m5^8, mix(value^m2, value^uint64(seed)^m1)))
+}
+
+func Hash128(value [16]byte, seed uintptr) uintptr {
+	a := binary.LittleEndian.Uint64(value[:8])
+	b := binary.LittleEndian.Uint64(value[8:])
+	return uintptr(mix(m5^16, mix(a^m2, b^uint64(seed)^m1)))
+}
+
+func MultiHash32(hashes []uintptr, values []uint32, seed uintptr) {
+	MultiHashUint32Array(hashes, sparse.MakeUint32Array(values), seed)
+}
+
+func MultiHash64(hashes []uintptr, values []uint64, seed uintptr) {
+	MultiHashUint64Array(hashes, sparse.MakeUint64Array(values), seed)
+}
+
+func MultiHash128(hashes []uintptr, values [][16]byte, seed uintptr) {
+	MultiHashUint128Array(hashes, sparse.MakeUint128Array(values), seed)
+}
diff --git a/hashprobe/wyhash/wyhash_amd64.go b/hashprobe/wyhash/wyhash_amd64.go
new file mode 100644
index 0000000..262ef41
--- /dev/null
+++ b/hashprobe/wyhash/wyhash_amd64.go
@@ -0,0 +1,14 @@
+//go:build !purego
+
+package wyhash
+
+import "github.com/segmentio/parquet-go/sparse"
+
+//go:noescape
+func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr)
+
+//go:noescape
+func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr)
+
+//go:noescape
+func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr)
diff --git a/hashprobe/wyhash/wyhash_amd64.s b/hashprobe/wyhash/wyhash_amd64.s
new file mode 100644
index 0000000..7b99879
--- /dev/null
+++ b/hashprobe/wyhash/wyhash_amd64.s
@@ -0,0 +1,118 @@
+//go:build !purego
+
+#include "textflag.h"
+
+#define m1 0xa0761d6478bd642f
+#define m2 0xe7037ed1a0b428db
+#define m3 0x8ebc6af09c88c6e3
+#define m4 0x589965cc75374cc3
+#define m5 0x1d8e4e27c47d124f
+
+// func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr)
+TEXT ·MultiHashUint32Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), R12
+    MOVQ values_array_ptr+24(FP), R13
+    MOVQ values_array_len+32(FP), R14
+    MOVQ values_array_off+40(FP), R15
+    MOVQ seed+48(FP), R11
+
+    MOVQ $m1, R8
+    MOVQ $m2, R9
+    MOVQ $m5^4, R10
+    XORQ R11, R8
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVL (R13), AX
+    MOVQ R8, BX
+
+    XORQ AX, BX
+    XORQ R9, AX
+
+    MULQ BX
+    XORQ DX, AX
+
+    MULQ R10
+    XORQ DX, AX
+
+    MOVQ AX, (R12)(SI*8)
+    INCQ SI
+    ADDQ R15, R13
+test:
+    CMPQ SI, R14
+    JNE loop
+    RET
+
+// func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr)
+TEXT ·MultiHashUint64Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), R12
+    MOVQ values_array_ptr+24(FP), R13
+    MOVQ values_array_len+32(FP), R14
+    MOVQ values_array_off+40(FP), R15
+    MOVQ seed+48(FP), R11
+
+    MOVQ $m1, R8
+    MOVQ $m2, R9
+    MOVQ $m5^8, R10
+    XORQ R11, R8
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ (R13), AX
+    MOVQ R8, BX
+
+    XORQ AX, BX
+    XORQ R9, AX
+
+    MULQ BX
+    XORQ DX, AX
+
+    MULQ R10
+    XORQ DX, AX
+
+    MOVQ AX, (R12)(SI*8)
+    INCQ SI
+    ADDQ R15, R13
+test:
+    CMPQ SI, R14
+    JNE loop
+    RET
+
+// func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr)
+TEXT ·MultiHashUint128Array(SB), NOSPLIT, $0-56
+    MOVQ hashes_base+0(FP), R12
+    MOVQ values_array_ptr+24(FP), R13
+    MOVQ values_array_len+32(FP), R14
+    MOVQ values_array_off+40(FP), R15
+    MOVQ seed+48(FP), R11
+
+    MOVQ $m1, R8
+    MOVQ $m2, R9
+    MOVQ $m5^16, R10
+    XORQ R11, R8
+
+    XORQ SI, SI
+    JMP test
+loop:
+    MOVQ 0(R13), AX
+    MOVQ 8(R13), DX
+    MOVQ R8, BX
+
+    XORQ DX, BX
+    XORQ R9, AX
+
+    MULQ BX
+    XORQ DX, AX
+
+    MULQ R10
+    XORQ DX, AX
+
+    MOVQ AX, (R12)(SI*8)
+    INCQ SI
+    ADDQ R15, R13
+test:
+    CMPQ SI, R14
+    JNE loop
+    RET
diff --git a/hashprobe/wyhash/wyhash_purego.go b/hashprobe/wyhash/wyhash_purego.go
new file mode 100644
index 0000000..c2ad707
--- /dev/null
+++ b/hashprobe/wyhash/wyhash_purego.go
@@ -0,0 +1,23 @@
+//go:build purego || !amd64
+
+package wyhash
+
+import "github.com/segmentio/parquet-go/sparse"
+
+func MultiHashUint32Array(hashes []uintptr, values sparse.Uint32Array, seed uintptr) {
+	for i := range hashes {
+		hashes[i] = Hash32(values.Index(i), seed)
+	}
+}
+
+func MultiHashUint64Array(hashes []uintptr, values sparse.Uint64Array, seed uintptr) {
+	for i := range hashes {
+		hashes[i] = Hash64(values.Index(i), seed)
+	}
+}
+
+func MultiHashUint128Array(hashes []uintptr, values sparse.Uint128Array, seed uintptr) {
+	for i := range hashes {
+		hashes[i] = Hash128(values.Index(i), seed)
+	}
+}
diff --git a/internal/bytealg/count_amd64.s b/internal/bytealg/count_amd64.s
index fe27189..d429614 100644
--- a/internal/bytealg/count_amd64.s
+++ b/internal/bytealg/count_amd64.s
@@ -6,7 +6,7 @@
 TEXT ·Count(SB), NOSPLIT, $0-40
     MOVQ data_base+0(FP), AX
     MOVQ data_len+8(FP), CX
-    MOVQ value+24(FP), BX
+    MOVB value+24(FP), BX
     MOVQ CX, DX // len
     ADDQ AX, CX // end
     XORQ SI, SI // count
diff --git a/null.go b/null.go
index 89090c3..1a0edf2 100644
--- a/null.go
+++ b/null.go
@@ -9,6 +9,7 @@ import (
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/internal/bytealg"
 	"github.com/segmentio/parquet-go/internal/unsafecast"
+	"github.com/segmentio/parquet-go/sparse"
 )
 
 // nullIndexFunc is the type of functions used to detect null values in rows.
@@ -18,12 +19,12 @@ import (
 //
 // 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)
+type nullIndexFunc func(bits []uint64, rows sparse.Array)
 
-func nullIndex[T comparable](bits []uint64, rows array, size, offset uintptr) {
+func nullIndex[T comparable](bits []uint64, rows sparse.Array) {
 	var zero T
-	for i := 0; i < rows.len; i++ {
-		v := *(*T)(rows.index(i, size, offset))
+	for i := 0; i < rows.Len(); i++ {
+		v := *(*T)(rows.Index(i))
 		if v != zero {
 			x := uint(i) / 64
 			y := uint(i) % 64
@@ -32,7 +33,7 @@ func nullIndex[T comparable](bits []uint64, rows array, size, offset uintptr) {
 	}
 }
 
-func nullIndexStruct(bits []uint64, rows array, size, offset uintptr) {
+func nullIndexStruct(bits []uint64, rows sparse.Array) {
 	bytealg.Broadcast(unsafecast.Slice[byte](bits), 0xFF)
 }
 
@@ -97,9 +98,9 @@ func nullIndexFuncOf(t reflect.Type) nullIndexFunc {
 }
 
 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))
+	return func(bits []uint64, rows sparse.Array) {
+		for i := 0; i < rows.Len(); i++ {
+			p := (*byte)(rows.Index(i))
 			b := unsafe.Slice(p, n)
 			if !isZero(b) {
 				x := uint(i) / 64
diff --git a/null_amd64.go b/null_amd64.go
index 19ad35f..2c80e25 100644
--- a/null_amd64.go
+++ b/null_amd64.go
@@ -2,71 +2,73 @@
 
 package parquet
 
+import "github.com/segmentio/parquet-go/sparse"
+
 //go:noescape
-func nullIndex8bits(bits *uint64, rows array, size, offset uintptr)
+func nullIndex8(bits *uint64, rows sparse.Array)
 
 //go:noescape
-func nullIndex32bits(bits *uint64, rows array, size, offset uintptr)
+func nullIndex32(bits *uint64, rows sparse.Array)
 
 //go:noescape
-func nullIndex64bits(bits *uint64, rows array, size, offset uintptr)
+func nullIndex64(bits *uint64, rows sparse.Array)
 
 //go:noescape
-func nullIndex128bits(bits *uint64, rows array, size, offset uintptr)
+func nullIndex128(bits *uint64, rows sparse.Array)
 
-func nullIndexBool(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex8bits(&bits[0], rows, size, offset)
+func nullIndexBool(bits []uint64, rows sparse.Array) {
+	nullIndex8(&bits[0], rows)
 }
 
-func nullIndexInt(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexInt(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexInt32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex32bits(&bits[0], rows, size, offset)
+func nullIndexInt32(bits []uint64, rows sparse.Array) {
+	nullIndex32(&bits[0], rows)
 }
 
-func nullIndexInt64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexInt64(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexUint(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexUint(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexUint32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex32bits(&bits[0], rows, size, offset)
+func nullIndexUint32(bits []uint64, rows sparse.Array) {
+	nullIndex32(&bits[0], rows)
 }
 
-func nullIndexUint64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexUint64(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexUint128(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex128bits(&bits[0], rows, size, offset)
+func nullIndexUint128(bits []uint64, rows sparse.Array) {
+	nullIndex128(&bits[0], rows)
 }
 
-func nullIndexFloat32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex32bits(&bits[0], rows, size, offset)
+func nullIndexFloat32(bits []uint64, rows sparse.Array) {
+	nullIndex32(&bits[0], rows)
 }
 
-func nullIndexFloat64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexFloat64(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexString(bits []uint64, rows array, size, offset uintptr) {
+func nullIndexString(bits []uint64, rows sparse.Array) {
 	// 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)
+	nullIndex64(&bits[0], rows.Offset(8))
 }
 
-func nullIndexSlice(bits []uint64, rows array, size, offset uintptr) {
+func nullIndexSlice(bits []uint64, rows sparse.Array) {
 	// 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)
+	nullIndex64(&bits[0], rows)
 }
 
-func nullIndexPointer(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex64bits(&bits[0], rows, size, offset)
+func nullIndexPointer(bits []uint64, rows sparse.Array) {
+	nullIndex64(&bits[0], rows)
 }
diff --git a/null_amd64.s b/null_amd64.s
index 20b85fb..52804d3 100644
--- a/null_amd64.s
+++ b/null_amd64.s
@@ -2,13 +2,12 @@
 
 #include "textflag.h"
 
-// func nullIndex8bits(bits *uint64, rows array, size, offset uintptr)
-TEXT ·nullIndex8bits(SB), NOSPLIT, $0-40
+// func nullIndex8(bits *uint64, rows sparse.Array)
+TEXT ·nullIndex8(SB), NOSPLIT, $0-32
     MOVQ bits+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), DI
-    MOVQ size+24(FP), DX
-    ADDQ offset+32(FP), BX
+    MOVQ rows_array_ptr+8(FP), BX
+    MOVQ rows_array_len+16(FP), DI
+    MOVQ rows_array_off+24(FP), DX
 
     MOVQ $1, CX
     XORQ SI, SI
@@ -33,13 +32,12 @@ next1x1:
 done:
     RET
 
-// func nullIndex32bits(bits *uint64, rows array, size, offset uintptr)
-TEXT ·nullIndex32bits(SB), NOSPLIT, $0-40
+// func nullIndex32(bits *uint64, rows sparse.Array)
+TEXT ·nullIndex32(SB), NOSPLIT, $0-32
     MOVQ bits+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), DI
-    MOVQ size+24(FP), DX
-    ADDQ offset+32(FP), BX
+    MOVQ rows_array_ptr+8(FP), BX
+    MOVQ rows_array_len+16(FP), DI
+    MOVQ rows_array_off+24(FP), DX
 
     MOVQ $1, CX
     XORQ SI, SI
@@ -57,7 +55,7 @@ TEXT ·nullIndex32bits(SB), NOSPLIT, $0-40
     SHRQ $3, R8
     SHLQ $3, R8
 
-    VPBROADCASTD size+24(FP), Y0
+    VPBROADCASTD rows_array_off+24(FP), Y0
     VPMULLD ·range0n8(SB), Y0, Y0
     VPCMPEQD Y1, Y1, Y1
     VPCMPEQD Y2, Y2, Y2
@@ -112,13 +110,12 @@ next1x4:
 done:
     RET
 
-// func nullIndex64bits(bits *uint64, rows array, size, offset uintptr)
-TEXT ·nullIndex64bits(SB), NOSPLIT, $0-40
+// func nullIndex64(bits *uint64, rows sparse.Array)
+TEXT ·nullIndex64(SB), NOSPLIT, $0-32
     MOVQ bits+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), DI
-    MOVQ size+24(FP), DX
-    ADDQ offset+32(FP), BX
+    MOVQ rows_array_ptr+8(FP), BX
+    MOVQ rows_array_len+16(FP), DI
+    MOVQ rows_array_off+24(FP), DX
 
     MOVQ $1, CX
     XORQ SI, SI
@@ -136,8 +133,8 @@ TEXT ·nullIndex64bits(SB), NOSPLIT, $0-40
     SHRQ $2, R8
     SHLQ $2, R8
 
-    VPBROADCASTQ size+24(FP), Y0
-    VPMULLD ·scale4x8(SB), Y0, Y0
+    VPBROADCASTQ rows_array_off+24(FP), Y0
+    VPMULLD scale4x8<>(SB), Y0, Y0
     VPCMPEQQ Y1, Y1, Y1
     VPCMPEQQ Y2, Y2, Y2
     VPXOR Y3, Y3, Y3
@@ -191,13 +188,18 @@ next1x8:
 done:
     RET
 
-// func nullIndex128bits(bits *uint64, rows array, size, offset uintptr)
-TEXT ·nullIndex128bits(SB), NOSPLIT, $0-40
+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
+
+// func nullIndex128(bits *uint64, rows sparse.Array)
+TEXT ·nullIndex128(SB), NOSPLIT, $0-32
     MOVQ bits+0(FP), AX
-    MOVQ rows_ptr+8(FP), BX
-    MOVQ rows_len+16(FP), DI
-    MOVQ size+24(FP), DX
-    ADDQ offset+32(FP), BX
+    MOVQ rows_array_ptr+8(FP), BX
+    MOVQ rows_array_len+16(FP), DI
+    MOVQ rows_array_off+24(FP), DX
 
     CMPQ DI, $0
     JE done
diff --git a/null_purego.go b/null_purego.go
index 06a63b9..b03c56b 100644
--- a/null_purego.go
+++ b/null_purego.go
@@ -2,53 +2,55 @@
 
 package parquet
 
-func nullIndexBool(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[bool](bits, rows, size, offset)
+import "github.com/segmentio/parquet-go/sparse"
+
+func nullIndexBool(bits []uint64, rows sparse.Array) {
+	nullIndex[bool](bits, rows)
 }
 
-func nullIndexInt(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[int](bits, rows, size, offset)
+func nullIndexInt(bits []uint64, rows sparse.Array) {
+	nullIndex[int](bits, rows)
 }
 
-func nullIndexInt32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[int32](bits, rows, size, offset)
+func nullIndexInt32(bits []uint64, rows sparse.Array) {
+	nullIndex[int32](bits, rows)
 }
 
-func nullIndexInt64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[int64](bits, rows, size, offset)
+func nullIndexInt64(bits []uint64, rows sparse.Array) {
+	nullIndex[int64](bits, rows)
 }
 
-func nullIndexUint(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[uint](bits, rows, size, offset)
+func nullIndexUint(bits []uint64, rows sparse.Array) {
+	nullIndex[uint](bits, rows)
 }
 
-func nullIndexUint32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[uint32](bits, rows, size, offset)
+func nullIndexUint32(bits []uint64, rows sparse.Array) {
+	nullIndex[uint32](bits, rows)
 }
 
-func nullIndexUint64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[uint64](bits, rows, size, offset)
+func nullIndexUint64(bits []uint64, rows sparse.Array) {
+	nullIndex[uint64](bits, rows)
 }
 
-func nullIndexUint128(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[[16]byte](bits, rows, size, offset)
+func nullIndexUint128(bits []uint64, rows sparse.Array) {
+	nullIndex[[16]byte](bits, rows)
 }
 
-func nullIndexFloat32(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[float32](bits, rows, size, offset)
+func nullIndexFloat32(bits []uint64, rows sparse.Array) {
+	nullIndex[float32](bits, rows)
 }
 
-func nullIndexFloat64(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[float64](bits, rows, size, offset)
+func nullIndexFloat64(bits []uint64, rows sparse.Array) {
+	nullIndex[float64](bits, rows)
 }
 
-func nullIndexString(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[string](bits, rows, size, offset)
+func nullIndexString(bits []uint64, rows sparse.Array) {
+	nullIndex[string](bits, rows)
 }
 
-func nullIndexSlice(bits []uint64, rows array, size, offset uintptr) {
-	for i := 0; i < rows.len; i++ {
-		p := *(**struct{})(rows.index(i, size, offset))
+func nullIndexSlice(bits []uint64, rows sparse.Array) {
+	for i := 0; i < rows.Len(); i++ {
+		p := *(**struct{})(rows.Index(i))
 		b := uint64(0)
 		if p != nil {
 			b = 1
@@ -57,6 +59,6 @@ func nullIndexSlice(bits []uint64, rows array, size, offset uintptr) {
 	}
 }
 
-func nullIndexPointer(bits []uint64, rows array, size, offset uintptr) {
-	nullIndex[*struct{}](bits, rows, size, offset)
+func nullIndexPointer(bits []uint64, rows sparse.Array) {
+	nullIndex[*struct{}](bits, rows)
 }
diff --git a/page_max_amd64.s b/page_max_amd64.s
index 0c62bee..afac16c 100644
--- a/page_max_amd64.s
+++ b/page_max_amd64.s
@@ -565,17 +565,13 @@ loop16:
     //
     // This loop is also taken if the CPU has no support for AVX-512.
 loop:
-    MOVQ (AX), R8
-    MOVQ (BX), R9
-    BSWAPQ R8
-    BSWAPQ R9
+    MOVBEQQ (AX), R8
+    MOVBEQQ (BX), R9
     CMPQ R8, R9
     JA more
     JB next
-    MOVQ 8(AX), R8
-    MOVQ 8(BX), R9
-    BSWAPQ R8
-    BSWAPQ R9
+    MOVBEQQ 8(AX), R8
+    MOVBEQQ 8(BX), R9
     CMPQ R8, R9
     JAE next
 more:
diff --git a/page_min_amd64.s b/page_min_amd64.s
index 7bd2e3a..73e2bbb 100644
--- a/page_min_amd64.s
+++ b/page_min_amd64.s
@@ -558,19 +558,15 @@ loop16:
     // Unless the input was aligned on 256 bytes, we need to perform a few more
     // iterations on the remaining elements.
     //
-    // This loop is also taken if the CPU has no support for AVX-512.
+    // This loop is also take if the CPU has no support for AVX-512.
 loop:
-    MOVQ (AX), R8
-    MOVQ (BX), R9
-    BSWAPQ R8
-    BSWAPQ R9
+    MOVBEQQ (AX), R8
+    MOVBEQQ (BX), R9
     CMPQ R8, R9
     JB less
     JA next
-    MOVQ 8(AX), R8
-    MOVQ 8(BX), R9
-    BSWAPQ R8
-    BSWAPQ R9
+    MOVBEQQ 8(AX), R8
+    MOVBEQQ 8(BX), R9
     CMPQ R8, R9
     JAE next
 less:
diff --git a/parquet_amd64.go b/parquet_amd64.go
index b2a6868..6416557 100644
--- a/parquet_amd64.go
+++ b/parquet_amd64.go
@@ -4,12 +4,6 @@ package parquet
 
 import "golang.org/x/sys/cpu"
 
-const (
-	// For very short inputs we are better off staying in Go code; since the
-	// functions get inlined the abstractions have zero overhead then.
-	minLenAVX2 = 8
-)
-
 var (
 	// This variable is used in x86 assembly source files to gate the use of
 	// AVX2 instructions depending on whether the CPU supports it.
@@ -20,9 +14,5 @@ var (
 	// follwing instructions from the DQ set:
 	// * VPBROADCASTQ (with 64 bits source register)
 	// * VBROADCASTI64X2
-	hasAVX512MinMaxBE128 = hasAVX512F && cpu.X86.HasAVX512DQ
+	hasAVX512MinMaxBE128 = cpu.X86.HasAVX512F && cpu.X86.HasAVX512DQ
 )
-
-func optimize(n int) bool {
-	return n >= minLenAVX2 && hasAVX2
-}
diff --git a/parquet_go18.go b/parquet_go18.go
index 9ebf72b..fa086eb 100644
--- a/parquet_go18.go
+++ b/parquet_go18.go
@@ -29,7 +29,7 @@ func Read[T any](r io.ReaderAt, size int64, options ...ReaderOption) (rows []T,
 	reader := NewGenericReader[T](file, config)
 	n, err := reader.Read(rows)
 	if n < len(rows) && err == io.EOF {
-		rows, err = rows[:n], err
+		rows, err = rows[:n], nil
 	}
 	reader.Close()
 	return rows, err
diff --git a/parquet_purego.go b/parquet_purego.go
deleted file mode 100644
index d2a2b4e..0000000
--- a/parquet_purego.go
+++ /dev/null
@@ -1,5 +0,0 @@
-//go:build purego || !amd64
-
-package parquet
-
-func optimize(int) bool { return false }
diff --git a/sparse/array.go b/sparse/array.go
new file mode 100644
index 0000000..5e4208b
--- /dev/null
+++ b/sparse/array.go
@@ -0,0 +1,272 @@
+package sparse
+
+import "unsafe"
+
+type Array struct{ array }
+
+func UnsafeArray(base unsafe.Pointer, length int, offset uintptr) Array {
+	return Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Array) Len() int                   { return int(a.len) }
+func (a Array) Index(i int) unsafe.Pointer { return a.index(i) }
+func (a Array) Slice(i, j int) Array       { return Array{a.slice(i, j)} }
+func (a Array) Offset(off uintptr) Array   { return Array{a.offset(off)} }
+func (a Array) BoolArray() BoolArray       { return BoolArray{a.array} }
+func (a Array) Int8Array() Int8Array       { return Int8Array{a.array} }
+func (a Array) Int16Array() Int16Array     { return Int16Array{a.array} }
+func (a Array) Int32Array() Int32Array     { return Int32Array{a.array} }
+func (a Array) Int64Array() Int64Array     { return Int64Array{a.array} }
+func (a Array) Float32Array() Float32Array { return Float32Array{a.array} }
+func (a Array) Float64Array() Float64Array { return Float64Array{a.array} }
+func (a Array) Uint8Array() Uint8Array     { return Uint8Array{a.array} }
+func (a Array) Uint16Array() Uint16Array   { return Uint16Array{a.array} }
+func (a Array) Uint32Array() Uint32Array   { return Uint32Array{a.array} }
+func (a Array) Uint64Array() Uint64Array   { return Uint64Array{a.array} }
+func (a Array) Uint128Array() Uint128Array { return Uint128Array{a.array} }
+
+type array struct {
+	ptr unsafe.Pointer
+	len uintptr
+	off uintptr
+}
+
+func makeArray(base unsafe.Pointer, length, offset uintptr) array {
+	return array{ptr: base, len: length, off: offset}
+}
+
+func (a array) index(i int) unsafe.Pointer {
+	if uintptr(i) >= a.len {
+		panic("index out of bounds")
+	}
+	return unsafe.Add(a.ptr, a.off*uintptr(i))
+}
+
+func (a array) slice(i, j int) array {
+	if uintptr(i) > a.len || uintptr(j) > a.len || i > j {
+		panic("slice index out of bounds")
+	}
+	return array{
+		ptr: unsafe.Add(a.ptr, a.off*uintptr(i)),
+		len: uintptr(j - i),
+		off: a.off,
+	}
+}
+
+func (a array) offset(off uintptr) array {
+	return array{
+		ptr: unsafe.Add(a.ptr, off),
+		len: a.len,
+		off: a.off,
+	}
+}
+
+type BoolArray struct{ array }
+
+func MakeBoolArray(values []bool) BoolArray {
+	return BoolArray{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 1)}
+}
+
+func UnsafeBoolArray(base unsafe.Pointer, length int, offset uintptr) BoolArray {
+	return BoolArray{makeArray(base, uintptr(length), offset)}
+}
+
+func (a BoolArray) Len() int                 { return int(a.len) }
+func (a BoolArray) Index(i int) bool         { return *(*byte)(a.index(i)) != 0 }
+func (a BoolArray) Slice(i, j int) BoolArray { return BoolArray{a.slice(i, j)} }
+func (a BoolArray) Uint8Array() Uint8Array   { return Uint8Array{a.array} }
+func (a BoolArray) UnsafeArray() Array       { return Array{a.array} }
+
+type Int8Array struct{ array }
+
+func MakeInt8Array(values []int8) Int8Array {
+	return Int8Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeInt8Array(base unsafe.Pointer, length int, offset uintptr) Int8Array {
+	return Int8Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Int8Array) Len() int                 { return int(a.len) }
+func (a Int8Array) Index(i int) int8         { return *(*int8)(a.index(i)) }
+func (a Int8Array) Slice(i, j int) Int8Array { return Int8Array{a.slice(i, j)} }
+func (a Int8Array) Uint8Array() Uint8Array   { return Uint8Array{a.array} }
+func (a Int8Array) UnsafeArray() Array       { return Array{a.array} }
+
+type Int16Array struct{ array }
+
+func MakeInt16Array(values []int16) Int16Array {
+	return Int16Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeInt16Array(base unsafe.Pointer, length int, offset uintptr) Int16Array {
+	return Int16Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Int16Array) Len() int                  { return int(a.len) }
+func (a Int16Array) Index(i int) int16         { return *(*int16)(a.index(i)) }
+func (a Int16Array) Slice(i, j int) Int16Array { return Int16Array{a.slice(i, j)} }
+func (a Int16Array) Int8Array() Int8Array      { return Int8Array{a.array} }
+func (a Int16Array) Uint8Array() Uint8Array    { return Uint8Array{a.array} }
+func (a Int16Array) Uint16Array() Uint16Array  { return Uint16Array{a.array} }
+func (a Int16Array) UnsafeArray() Array        { return Array{a.array} }
+
+type Int32Array struct{ array }
+
+func MakeInt32Array(values []int32) Int32Array {
+	return Int32Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 4)}
+}
+
+func UnsafeInt32Array(base unsafe.Pointer, length int, offset uintptr) Int32Array {
+	return Int32Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Int32Array) Len() int                  { return int(a.len) }
+func (a Int32Array) Index(i int) int32         { return *(*int32)(a.index(i)) }
+func (a Int32Array) Slice(i, j int) Int32Array { return Int32Array{a.slice(i, j)} }
+func (a Int32Array) Int8Array() Int8Array      { return Int8Array{a.array} }
+func (a Int32Array) Int16Array() Int16Array    { return Int16Array{a.array} }
+func (a Int32Array) Uint8Array() Uint8Array    { return Uint8Array{a.array} }
+func (a Int32Array) Uint16Array() Uint16Array  { return Uint16Array{a.array} }
+func (a Int32Array) Uint32Array() Uint32Array  { return Uint32Array{a.array} }
+func (a Int32Array) UnsafeArray() Array        { return Array{a.array} }
+
+type Int64Array struct{ array }
+
+func MakeInt64Array(values []int64) Int64Array {
+	return Int64Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeInt64Array(base unsafe.Pointer, length int, offset uintptr) Int64Array {
+	return Int64Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Int64Array) Len() int                  { return int(a.len) }
+func (a Int64Array) Index(i int) int64         { return *(*int64)(a.index(i)) }
+func (a Int64Array) Slice(i, j int) Int64Array { return Int64Array{a.slice(i, j)} }
+func (a Int64Array) Int8Array() Int8Array      { return Int8Array{a.array} }
+func (a Int64Array) Int16Array() Int16Array    { return Int16Array{a.array} }
+func (a Int64Array) Int32Array() Int32Array    { return Int32Array{a.array} }
+func (a Int64Array) Uint8Array() Uint8Array    { return Uint8Array{a.array} }
+func (a Int64Array) Uint16Array() Uint16Array  { return Uint16Array{a.array} }
+func (a Int64Array) Uint32Array() Uint32Array  { return Uint32Array{a.array} }
+func (a Int64Array) Uint64Array() Uint64Array  { return Uint64Array{a.array} }
+func (a Int64Array) UnsafeArray() Array        { return Array{a.array} }
+
+type Float32Array struct{ array }
+
+func MakeFloat32Array(values []float32) Float32Array {
+	return Float32Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 4)}
+}
+
+func UnsafeFloat32Array(base unsafe.Pointer, length int, offset uintptr) Float32Array {
+	return Float32Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Float32Array) Len() int                    { return int(a.len) }
+func (a Float32Array) Index(i int) float32         { return *(*float32)(a.index(i)) }
+func (a Float32Array) Slice(i, j int) Float32Array { return Float32Array{a.slice(i, j)} }
+func (a Float32Array) Array() Array                { return Array{a.array} }
+func (a Float32Array) Uint32Array() Uint32Array    { return Uint32Array{a.array} }
+func (a Float32Array) UnsafeArray() Array          { return Array{a.array} }
+
+type Float64Array struct{ array }
+
+func MakeFloat64Array(values []float64) Float64Array {
+	return Float64Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeFloat64Array(base unsafe.Pointer, length int, offset uintptr) Float64Array {
+	return Float64Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Float64Array) Len() int                    { return int(a.len) }
+func (a Float64Array) Index(i int) float64         { return *(*float64)(a.index(i)) }
+func (a Float64Array) Slice(i, j int) Float64Array { return Float64Array{a.slice(i, j)} }
+func (a Float64Array) Uint64Array() Uint64Array    { return Uint64Array{a.array} }
+func (a Float64Array) UnsafeArray() Array          { return Array{a.array} }
+
+type Uint8Array struct{ array }
+
+func MakeUint8Array(values []uint8) Uint8Array {
+	return Uint8Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeUint8Array(base unsafe.Pointer, length int, offset uintptr) Uint8Array {
+	return Uint8Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Uint8Array) Len() int                  { return int(a.len) }
+func (a Uint8Array) Index(i int) uint8         { return *(*uint8)(a.index(i)) }
+func (a Uint8Array) Slice(i, j int) Uint8Array { return Uint8Array{a.slice(i, j)} }
+func (a Uint8Array) UnsafeArray() Array        { return Array{a.array} }
+
+type Uint16Array struct{ array }
+
+func MakeUint16Array(values []uint16) Uint16Array {
+	return Uint16Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeUint16Array(base unsafe.Pointer, length int, offset uintptr) Uint16Array {
+	return Uint16Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Uint16Array) Len() int                   { return int(a.len) }
+func (a Uint16Array) Index(i int) uint16         { return *(*uint16)(a.index(i)) }
+func (a Uint16Array) Slice(i, j int) Uint16Array { return Uint16Array{a.slice(i, j)} }
+func (a Uint16Array) Uint8Array() Uint8Array     { return Uint8Array{a.array} }
+func (a Uint16Array) UnsafeArray() Array         { return Array{a.array} }
+
+type Uint32Array struct{ array }
+
+func MakeUint32Array(values []uint32) Uint32Array {
+	return Uint32Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 4)}
+}
+
+func UnsafeUint32Array(base unsafe.Pointer, length int, offset uintptr) Uint32Array {
+	return Uint32Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Uint32Array) Len() int                   { return int(a.len) }
+func (a Uint32Array) Index(i int) uint32         { return *(*uint32)(a.index(i)) }
+func (a Uint32Array) Slice(i, j int) Uint32Array { return Uint32Array{a.slice(i, j)} }
+func (a Uint32Array) Uint8Array() Uint8Array     { return Uint8Array{a.array} }
+func (a Uint32Array) Uint16Array() Uint16Array   { return Uint16Array{a.array} }
+func (a Uint32Array) UnsafeArray() Array         { return Array{a.array} }
+
+type Uint64Array struct{ array }
+
+func MakeUint64Array(values []uint64) Uint64Array {
+	return Uint64Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 8)}
+}
+
+func UnsafeUint64Array(base unsafe.Pointer, length int, offset uintptr) Uint64Array {
+	return Uint64Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Uint64Array) Len() int                   { return int(a.len) }
+func (a Uint64Array) Index(i int) uint64         { return *(*uint64)(a.index(i)) }
+func (a Uint64Array) Slice(i, j int) Uint64Array { return Uint64Array{a.slice(i, j)} }
+func (a Uint64Array) Uint8Array() Uint8Array     { return Uint8Array{a.array} }
+func (a Uint64Array) Uint16Array() Uint16Array   { return Uint16Array{a.array} }
+func (a Uint64Array) Uint32Array() Uint32Array   { return Uint32Array{a.array} }
+func (a Uint64Array) UnsafeArray() Array         { return Array{a.array} }
+
+type Uint128Array struct{ array }
+
+func MakeUint128Array(values [][16]byte) Uint128Array {
+	return Uint128Array{makeArray(*(*unsafe.Pointer)(unsafe.Pointer(&values)), uintptr(len(values)), 16)}
+}
+
+func UnsafeUint128Array(base unsafe.Pointer, length int, offset uintptr) Uint128Array {
+	return Uint128Array{makeArray(base, uintptr(length), offset)}
+}
+
+func (a Uint128Array) Len() int                    { return int(a.len) }
+func (a Uint128Array) Index(i int) [16]byte        { return *(*[16]byte)(a.index(i)) }
+func (a Uint128Array) Slice(i, j int) Uint128Array { return Uint128Array{a.slice(i, j)} }
+func (a Uint128Array) Uint8Array() Uint8Array      { return Uint8Array{a.array} }
+func (a Uint128Array) Uint16Array() Uint16Array    { return Uint16Array{a.array} }
+func (a Uint128Array) Uint32Array() Uint32Array    { return Uint32Array{a.array} }
+func (a Uint128Array) Uint64Array() Uint64Array    { return Uint64Array{a.array} }
+func (a Uint128Array) UnsafeArray() Array          { return Array{a.array} }
diff --git a/sparse/gather.go b/sparse/gather.go
new file mode 100644
index 0000000..ac83897
--- /dev/null
+++ b/sparse/gather.go
@@ -0,0 +1,34 @@
+package sparse
+
+import "unsafe"
+
+func GatherInt32(dst []int32, src Int32Array) int {
+	return GatherUint32(*(*[]uint32)(unsafe.Pointer(&dst)), src.Uint32Array())
+}
+
+func GatherInt64(dst []int64, src Int64Array) int {
+	return GatherUint64(*(*[]uint64)(unsafe.Pointer(&dst)), src.Uint64Array())
+}
+
+func GatherFloat32(dst []float32, src Float32Array) int {
+	return GatherUint32(*(*[]uint32)(unsafe.Pointer(&dst)), src.Uint32Array())
+}
+
+func GatherFloat64(dst []float64, src Float64Array) int {
+	return GatherUint64(*(*[]uint64)(unsafe.Pointer(&dst)), src.Uint64Array())
+}
+
+func GatherBits(dst []byte, src Uint8Array) int { return gatherBits(dst, src) }
+
+func GatherUint32(dst []uint32, src Uint32Array) int { return gather32(dst, src) }
+
+func GatherUint64(dst []uint64, src Uint64Array) int { return gather64(dst, src) }
+
+func GatherUint128(dst [][16]byte, src Uint128Array) int { return gather128(dst, src) }
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
diff --git a/sparse/gather_amd64.go b/sparse/gather_amd64.go
new file mode 100644
index 0000000..8b24cee
--- /dev/null
+++ b/sparse/gather_amd64.go
@@ -0,0 +1,83 @@
+//go:build !purego
+
+package sparse
+
+import "golang.org/x/sys/cpu"
+
+func gatherBits(dst []byte, src Uint8Array) int {
+	n := min(len(dst)*8, src.Len())
+	i := 0
+
+	if n >= 8 {
+		i = (n / 8) * 8
+		// Make sure `offset` is at least 4 bytes, otherwise VPGATHERDD may read
+		// data beyond the end of the program memory and trigger a fault.
+		//
+		// If the boolean values do not have enough padding we must fallback to
+		// the scalar algorithm to be able to load single bytes from memory.
+		if src.off >= 4 && cpu.X86.HasAVX2 {
+			gatherBitsAVX2(dst, src.Slice(0, i))
+		} else {
+			gatherBitsDefault(dst, src.Slice(0, i))
+		}
+	}
+
+	for i < n {
+		x := i / 8
+		y := i % 8
+		b := src.Index(i)
+		dst[x] = ((b & 1) << y) | (dst[x] & ^(1 << y))
+		i++
+	}
+
+	return n
+}
+
+func gather32(dst []uint32, src Uint32Array) int {
+	n := min(len(dst), src.Len())
+	i := 0
+
+	if n >= 16 && cpu.X86.HasAVX2 {
+		i = (n / 8) * 8
+		gather32AVX2(dst[:i:i], src)
+	}
+
+	for i < n {
+		dst[i] = src.Index(i)
+		i++
+	}
+
+	return n
+}
+
+func gather64(dst []uint64, src Uint64Array) int {
+	n := min(len(dst), src.Len())
+	i := 0
+
+	if n >= 8 && cpu.X86.HasAVX2 {
+		i = (n / 4) * 4
+		gather64AVX2(dst[:i:i], src)
+	}
+
+	for i < n {
+		dst[i] = src.Index(i)
+		i++
+	}
+
+	return n
+}
+
+//go:noescape
+func gatherBitsAVX2(dst []byte, src Uint8Array)
+
+//go:noescape
+func gatherBitsDefault(dst []byte, src Uint8Array)
+
+//go:noescape
+func gather32AVX2(dst []uint32, src Uint32Array)
+
+//go:noescape
+func gather64AVX2(dst []uint64, src Uint64Array)
+
+//go:noescape
+func gather128(dst [][16]byte, src Uint128Array) int
diff --git a/sparse/gather_amd64.s b/sparse/gather_amd64.s
new file mode 100644
index 0000000..aef4fee
--- /dev/null
+++ b/sparse/gather_amd64.s
@@ -0,0 +1,193 @@
+//go:build !purego
+
+#include "textflag.h"
+
+// func gatherBitsAVX2(dst []byte, src Uint8Array)
+TEXT ·gatherBitsAVX2(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ src_array_ptr+24(FP), BX
+    MOVQ src_array_len+32(FP), CX
+    MOVQ src_array_off+40(FP), DX
+    XORQ SI, SI
+    SHRQ $3, CX
+
+    VPBROADCASTD src_array_off+40(FP), Y0
+    VPMULLD range0n7x32<>(SB), Y0, Y0
+    VPCMPEQD Y1, Y1, Y1
+    VPCMPEQD Y2, Y2, Y2
+avx2loop:
+    VPGATHERDD Y1, (BX)(Y0*1), Y3
+    VMOVDQU Y2, Y1
+    VPSLLD $31, Y3, Y3
+    VMOVMSKPS Y3, DI
+
+    MOVB DI, (AX)(SI*1)
+
+    LEAQ (BX)(DX*8), BX
+    INCQ SI
+    CMPQ SI, CX
+    JNE avx2loop
+    VZEROUPPER
+    RET
+
+// func gatherBitsDefault(dst []byte, src Uint8Array)
+TEXT ·gatherBitsDefault(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ src_array_ptr+24(FP), BX
+    MOVQ src_array_len+32(FP), CX
+    MOVQ src_array_off+40(FP), DX
+    XORQ SI, SI
+    SHRQ $3, CX
+loop:
+    LEAQ (BX)(DX*2), DI
+    MOVBQZX (BX), R8
+    MOVBQZX (BX)(DX*1), R9
+    MOVBQZX (DI), R10
+    MOVBQZX (DI)(DX*1), R11
+    LEAQ (BX)(DX*4), BX
+    LEAQ (DI)(DX*4), DI
+    MOVBQZX (BX), R12
+    MOVBQZX (BX)(DX*1), R13
+    MOVBQZX (DI), R14
+    MOVBQZX (DI)(DX*1), R15
+    LEAQ (BX)(DX*4), BX
+
+    ANDQ $1, R8
+    ANDQ $1, R9
+    ANDQ $1, R10
+    ANDQ $1, R11
+    ANDQ $1, R12
+    ANDQ $1, R13
+    ANDQ $1, R14
+    ANDQ $1, R15
+
+    SHLQ $1, R9
+    SHLQ $2, R10
+    SHLQ $3, R11
+    SHLQ $4, R12
+    SHLQ $5, R13
+    SHLQ $6, R14
+    SHLQ $7, R15
+
+    ORQ R9, R8
+    ORQ R11, R10
+    ORQ R13, R12
+    ORQ R15, R14
+    ORQ R10, R8
+    ORQ R12, R8
+    ORQ R14, R8
+
+    MOVB R8, (AX)(SI*1)
+
+    INCQ SI
+    CMPQ SI, CX
+    JNE loop
+    RET
+
+// func gather32AVX2(dst []uint32, src Uint32Array)
+TEXT ·gather32AVX2(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ dst_len+8(FP), CX
+    MOVQ src_array_ptr+24(FP), BX
+    MOVQ src_array_off+40(FP), DX
+    XORQ SI, SI
+
+    VPBROADCASTD src_array_off+40(FP), Y0
+    VPMULLD range0n7x32<>(SB), Y0, Y0
+    VPCMPEQD Y1, Y1, Y1
+    VPCMPEQD Y2, Y2, Y2
+loop:
+    VPGATHERDD Y1, (BX)(Y0*1), Y3
+    VMOVDQU Y3, (AX)(SI*4)
+    VMOVDQU Y2, Y1
+
+    LEAQ (BX)(DX*8), BX
+    ADDQ $8, SI
+    CMPQ SI, CX
+    JNE loop
+    VZEROUPPER
+    RET
+
+// func gather64AVX2(dst []uint64, src Uint64Array)
+TEXT ·gather64AVX2(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ dst_len+8(FP), CX
+    MOVQ src_array_ptr+24(FP), BX
+    MOVQ src_array_off+40(FP), DX
+    XORQ SI, SI
+
+    VPBROADCASTQ src_array_off+40(FP), Y0
+    VPMULLD range0n3x64<>(SB), Y0, Y0
+    VPCMPEQQ Y1, Y1, Y1
+    VPCMPEQQ Y2, Y2, Y2
+loop:
+    VPGATHERQQ Y1, (BX)(Y0*1), Y3
+    VMOVDQU Y3, (AX)(SI*8)
+    VMOVDQU Y2, Y1
+
+    LEAQ (BX)(DX*4), BX
+    ADDQ $4, SI
+    CMPQ SI, CX
+    JNE loop
+    VZEROUPPER
+    RET
+
+// func gather128(dst [][16]byte, src Uint128Array) int
+TEXT ·gather128(SB), NOSPLIT, $0-56
+    MOVQ dst_base+0(FP), AX
+    MOVQ dst_len+8(FP), CX
+    MOVQ src_array_ptr+24(FP), BX
+    MOVQ src_array_len+32(FP), DI
+    MOVQ src_array_off+40(FP), DX
+    XORQ SI, SI
+
+    CMPQ DI, CX
+    CMOVQLT DI, CX
+
+    CMPQ CX, $0
+    JE done
+
+    CMPQ CX, $1
+    JE tail
+
+    XORQ SI, SI
+    MOVQ CX, DI
+    SHRQ $1, DI
+    SHLQ $1, DI
+loop:
+    MOVOU (BX), X0
+    MOVOU (BX)(DX*1), X1
+
+    MOVOU X0, (AX)
+    MOVOU X1, 16(AX)
+
+    LEAQ (BX)(DX*2), BX
+    ADDQ $32, AX
+    ADDQ $2, SI
+    CMPQ SI, DI
+    JNE loop
+
+    CMPQ SI, CX
+    JE done
+tail:
+    MOVOU (BX), X0
+    MOVOU X0, (AX)
+done:
+    MOVQ CX, ret+48(FP)
+    RET
+
+GLOBL range0n3x64<>(SB), RODATA|NOPTR, $32
+DATA range0n3x64<>+0(SB)/8,  $0
+DATA range0n3x64<>+8(SB)/8,  $1
+DATA range0n3x64<>+16(SB)/8, $2
+DATA range0n3x64<>+24(SB)/8, $3
+
+GLOBL range0n7x32<>(SB), RODATA|NOPTR, $32
+DATA range0n7x32<>+0(SB)/4,  $0
+DATA range0n7x32<>+4(SB)/4,  $1
+DATA range0n7x32<>+8(SB)/4,  $2
+DATA range0n7x32<>+12(SB)/4, $3
+DATA range0n7x32<>+16(SB)/4, $4
+DATA range0n7x32<>+20(SB)/4, $5
+DATA range0n7x32<>+24(SB)/4, $6
+DATA range0n7x32<>+28(SB)/4, $7
diff --git a/sparse/gather_purego.go b/sparse/gather_purego.go
new file mode 100644
index 0000000..2f25c44
--- /dev/null
+++ b/sparse/gather_purego.go
@@ -0,0 +1,72 @@
+//go:build purego || !amd64
+
+package sparse
+
+func gatherBits(dst []byte, src Uint8Array) int {
+	n := min(len(dst)*8, src.Len())
+	i := 0
+
+	if k := (n / 8) * 8; k > 0 {
+		for j := 0; i < k; j++ {
+			b0 := src.Index(i + 0)
+			b1 := src.Index(i + 1)
+			b2 := src.Index(i + 2)
+			b3 := src.Index(i + 3)
+			b4 := src.Index(i + 4)
+			b5 := src.Index(i + 5)
+			b6 := src.Index(i + 6)
+			b7 := src.Index(i + 7)
+
+			dst[j] = (b0 & 1) |
+				((b1 & 1) << 1) |
+				((b2 & 1) << 2) |
+				((b3 & 1) << 3) |
+				((b4 & 1) << 4) |
+				((b5 & 1) << 5) |
+				((b6 & 1) << 6) |
+				((b7 & 1) << 7)
+
+			i += 8
+		}
+	}
+
+	for i < n {
+		x := i / 8
+		y := i % 8
+		b := src.Index(i)
+		dst[x] = ((b & 1) << y) | (dst[x] & ^(1 << y))
+		i++
+	}
+
+	return n
+}
+
+func gather32(dst []uint32, src Uint32Array) int {
+	n := min(len(dst), src.Len())
+
+	for i := range dst[:n] {
+		dst[i] = src.Index(i)
+	}
+
+	return n
+}
+
+func gather64(dst []uint64, src Uint64Array) int {
+	n := min(len(dst), src.Len())
+
+	for i := range dst[:n] {
+		dst[i] = src.Index(i)
+	}
+
+	return n
+}
+
+func gather128(dst [][16]byte, src Uint128Array) int {
+	n := min(len(dst), src.Len())
+
+	for i := range dst[:n] {
+		dst[i] = src.Index(i)
+	}
+
+	return n
+}
diff --git a/sparse/sparse.go b/sparse/sparse.go
new file mode 100644
index 0000000..b2680bf
--- /dev/null
+++ b/sparse/sparse.go
@@ -0,0 +1,20 @@
+// Package sparse contains abstractions to help work on arrays of values in
+// sparse memory locations.
+//
+// Conversion between array types is supported when converting integers to a
+// lower size (e.g. int32 to int16, or uint64 to uint8), or converting from
+// signed integers to unsigned. Float types can also be converted to unsigned
+// integers of the same size, in which case the conversion is similar to using
+// the standard library's math.Float32bits and math.Float64bits functions.
+//
+// All array types can be converted to a generic Array type that can be used to erase
+// type information and bypass type conversion rules. This conversion is similar
+// to using Go's unsafe package to bypass Go's type system and should usually be
+// avoided and a sign that the application is attempting to break type safety
+// boundaries.
+//
+// The package provides Gather* functions which retrieve values from sparse
+// arrays into contiguous memory buffers. On platforms that support it, these
+// operations are implemented using SIMD gather instructions (e.g. VPGATHER on
+// Intel CPUs).
+package sparse
diff --git a/type.go b/type.go
index d3923af..dd3bdf1 100644
--- a/type.go
+++ b/type.go
@@ -999,7 +999,7 @@ func JSON() Node { return Leaf(&jsonType{}) }
 
 type jsonType format.JsonType
 
-func (t *jsonType) String() string { return (*format.JsonType)(t).String() }
+func (t *jsonType) String() string { return (*jsonType)(t).String() }
 
 func (t *jsonType) Kind() Kind { return ByteArray }
 
diff --git a/value_amd64.go b/value_amd64.go
index f1c8d11..fbd6432 100644
--- a/value_amd64.go
+++ b/value_amd64.go
@@ -2,11 +2,13 @@
 
 package parquet
 
+import "golang.org/x/sys/cpu"
+
 //go:noescape
 func memsetValuesAVX2(values []Value, model Value, _ uint64)
 
 func memsetValues(values []Value, model Value) {
-	if hasAVX2 {
+	if cpu.X86.HasAVX2 {
 		memsetValuesAVX2(values, model, 0)
 	} else {
 		for i := range values {
diff --git a/writer_go18.go b/writer_go18.go
index 397f705..ca6e01b 100644
--- a/writer_go18.go
+++ b/writer_go18.go
@@ -106,7 +106,6 @@ func writeFuncOf[T any](t reflect.Type, schema *Schema) writeFunc[T] {
 }
 
 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) {
 		if w.columns == nil {
@@ -119,7 +118,7 @@ func makeWriteFunc[T any](t reflect.Type, schema *Schema) writeFunc[T] {
 				w.columns[i] = c.columnBuffer
 			}
 		}
-		err = writeRows(w.columns, makeArrayOf(rows), size, 0, columnLevels{})
+		err = writeRows(w.columns, makeArrayOf(rows), columnLevels{})
 		if err == nil {
 			n = len(rows)
 		}


Test patch:
diff --git a/dictionary_test.go b/dictionary_test.go
index c283112..4ab350b 100644
--- a/dictionary_test.go
+++ b/dictionary_test.go
@@ -1,6 +1,7 @@
 package parquet_test
 
 import (
+	"fmt"
 	"math/rand"
 	"testing"
 	"time"
@@ -24,15 +25,16 @@ var dictionaryTypes = [...]parquet.Type{
 
 func TestDictionary(t *testing.T) {
 	for _, typ := range dictionaryTypes {
-		t.Run(typ.String(), func(t *testing.T) {
-			testDictionary(t, typ)
-		})
+		for _, numValues := range []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 1e2, 1e3, 1e4} {
+			t.Run(fmt.Sprintf("%s/N=%d", typ, numValues), func(t *testing.T) {
+				testDictionary(t, typ, numValues)
+			})
+		}
 	}
 }
 
-func testDictionary(t *testing.T, typ parquet.Type) {
+func testDictionary(t *testing.T, typ parquet.Type, numValues int) {
 	const columnIndex = 1
-	const numValues = 500
 
 	dict := typ.NewDictionary(columnIndex, 0, nil)
 	values := make([]parquet.Value, numValues)
@@ -40,7 +42,7 @@ func testDictionary(t *testing.T, typ parquet.Type) {
 	lookups := make([]parquet.Value, numValues)
 
 	f := randValueFuncOf(typ)
-	r := rand.New(rand.NewSource(0))
+	r := rand.New(rand.NewSource(int64(numValues)))
 
 	for i := range values {
 		values[i] = f(r)
@@ -122,8 +124,6 @@ func testDictionary(t *testing.T, typ parquet.Type) {
 }
 
 func BenchmarkDictionary(b *testing.B) {
-	const numValues = 1000
-
 	tests := []struct {
 		scenario string
 		init     func(parquet.Dictionary, []int32, []parquet.Value)
@@ -149,34 +149,36 @@ func BenchmarkDictionary(b *testing.B) {
 		},
 	}
 
-	for _, test := range tests {
+	for i, test := range tests {
 		b.Run(test.scenario, func(b *testing.B) {
-			for _, typ := range dictionaryTypes {
-				dict := typ.NewDictionary(0, 0, make([]byte, 0, 4*numValues))
-				values := make([]parquet.Value, numValues)
+			for j, typ := range dictionaryTypes {
+				for _, numValues := range []int{1e2, 1e3, 1e4, 1e5, 1e6} {
+					dict := typ.NewDictionary(0, 0, make([]byte, 0, 4*numValues))
+					values := make([]parquet.Value, numValues)
 
-				f := randValueFuncOf(typ)
-				r := rand.New(rand.NewSource(0))
+					f := randValueFuncOf(typ)
+					r := rand.New(rand.NewSource(int64(i * j * numValues)))
 
-				for i := range values {
-					values[i] = f(r)
-				}
+					for i := range values {
+						values[i] = f(r)
+					}
 
-				indexes := make([]int32, len(values))
-				if test.init != nil {
-					test.init(dict, indexes, values)
-				}
+					indexes := make([]int32, len(values))
+					if test.init != nil {
+						test.init(dict, indexes, values)
+					}
 
-				b.Run(typ.String(), func(b *testing.B) {
-					start := time.Now()
+					b.Run(fmt.Sprintf("%s/N=%d", typ, numValues), func(b *testing.B) {
+						start := time.Now()
 
-					for i := 0; i < b.N; i++ {
-						test.test(dict, indexes, values)
-					}
+						for i := 0; i < b.N; i++ {
+							test.test(dict, indexes, values)
+						}
 
-					seconds := time.Since(start).Seconds()
-					b.ReportMetric(float64(numValues*b.N)/seconds, "value/s")
-				})
+						seconds := time.Since(start).Seconds()
+						b.ReportMetric(float64(numValues*b.N)/seconds, "value/s")
+					})
+				}
 			}
 		})
 	}
diff --git a/hashprobe/aeshash/aeshash_test.go b/hashprobe/aeshash/aeshash_test.go
new file mode 100644
index 0000000..7092c65
--- /dev/null
+++ b/hashprobe/aeshash/aeshash_test.go
@@ -0,0 +1,172 @@
+package aeshash
+
+import (
+	"encoding/binary"
+	"testing"
+	"time"
+	"unsafe"
+)
+
+//go:noescape
+//go:linkname runtime_memhash32 runtime.memhash32
+func runtime_memhash32(data unsafe.Pointer, seed uintptr) uintptr
+
+//go:noescape
+//go:linkname runtime_memhash64 runtime.memhash64
+func runtime_memhash64(data unsafe.Pointer, seed uintptr) uintptr
+
+//go:noescape
+//go:linkname runtime_memhash runtime.memhash
+func runtime_memhash(data unsafe.Pointer, seed, size uintptr) uintptr
+
+func memhash32(data uint32, seed uintptr) uintptr {
+	return runtime_memhash32(unsafe.Pointer(&data), seed)
+}
+
+func memhash64(data uint64, seed uintptr) uintptr {
+	return runtime_memhash64(unsafe.Pointer(&data), seed)
+}
+
+func memhash128(data [16]byte, seed uintptr) uintptr {
+	return runtime_memhash(unsafe.Pointer(&data), seed, 16)
+}
+
+func TestHash32(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	h0 := memhash32(42, 1)
+	h1 := Hash32(42, 1)
+
+	if h0 != h1 {
+		t.Errorf("want=%016x got=%016x", h0, h1)
+	}
+}
+
+func TestMultiHash32(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N]uint32{}
+	seed := uintptr(32)
+
+	for i := range values {
+		values[i] = uint32(i)
+	}
+
+	MultiHash32(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash32(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%016x got=%016x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func TestHash64(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	h0 := memhash64(42, 1)
+	h1 := Hash64(42, 1)
+
+	if h0 != h1 {
+		t.Errorf("want=%016x got=%016x", h0, h1)
+	}
+}
+
+func TestMultiHash64(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N]uint64{}
+	seed := uintptr(64)
+
+	for i := range values {
+		values[i] = uint64(i)
+	}
+
+	MultiHash64(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash64(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%016x got=%016x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func BenchmarkMultiHash64(b *testing.B) {
+	if !Enabled() {
+		b.Skip("AES hash not supported on this platform")
+	}
+
+	hashes := [512]uintptr{}
+	values := [512]uint64{}
+	b.SetBytes(8 * int64(len(hashes)))
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		MultiHash64(hashes[:], values[:], seed)
+		return len(hashes)
+	})
+}
+
+func TestHash128(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	h0 := memhash128([16]byte{0: 42}, 1)
+	h1 := Hash128([16]byte{0: 42}, 1)
+
+	if h0 != h1 {
+		t.Errorf("want=%016x got=%016x", h0, h1)
+	}
+}
+
+func TestMultiHash128(t *testing.T) {
+	if !Enabled() {
+		t.Skip("AES hash not supported on this platform")
+	}
+
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N][16]byte{}
+	seed := uintptr(128)
+
+	for i := range values {
+		binary.LittleEndian.PutUint64(values[i][:8], uint64(i))
+	}
+
+	MultiHash128(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash128(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%016x got=%016x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func benchmarkHashThroughput(b *testing.B, f func(seed uintptr) int) {
+	hashes := int64(0)
+	start := time.Now()
+
+	for i := 0; i < b.N; i++ {
+		hashes += int64(f(uintptr(i)))
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(hashes)/seconds, "hash/s")
+}
diff --git a/hashprobe/hashprobe_test.go b/hashprobe/hashprobe_test.go
new file mode 100644
index 0000000..cd635d7
--- /dev/null
+++ b/hashprobe/hashprobe_test.go
@@ -0,0 +1,447 @@
+package hashprobe
+
+import (
+	"encoding/binary"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+	"unsafe"
+)
+
+func TestTable32GroupSize(t *testing.T) {
+	if n := unsafe.Sizeof(table32Group{}); n != 64 {
+		t.Errorf("size of 32 bit table group is not 64 bytes: %d", n)
+	}
+}
+
+func TestUint32TableProbeOneByOne(t *testing.T) {
+	const N = 500
+	table := NewUint32Table(0, 0.9)
+
+	for n := 0; n < 2; n++ {
+		// Do two passes, both should behave the same.
+		for i := 1; i <= N; i++ {
+			k := [1]uint32{}
+			v := [1]int32{}
+
+			k[0] = uint32(i)
+			table.Probe(k[:], v[:])
+
+			if v[0] != int32(i-1) {
+				t.Errorf("wrong value probed for key=%d: want=%d got=%d", i, i-1, v[0])
+			}
+		}
+	}
+}
+
+func TestUint32TableProbeBulk(t *testing.T) {
+	const N = 999
+	table := NewUint32Table(0, 0.9)
+
+	k := make([]uint32, N)
+	v := make([]int32, N)
+
+	for i := range k {
+		k[i] = uint32(i)
+	}
+
+	for n := 0; n < 2; n++ {
+		table.Probe(k, v)
+
+		for i := range v {
+			if v[i] != int32(i) {
+				t.Errorf("wrong value probed for key=%d: want=%d got=%d", k[i], i, v[i])
+			}
+		}
+
+		if t.Failed() {
+			break
+		}
+
+		for i := range v {
+			v[i] = 0
+		}
+	}
+}
+
+func TestTable64GroupSize(t *testing.T) {
+	if n := unsafe.Sizeof(table64Group{}); n != 64 {
+		t.Errorf("size of 64 bit table group is not 64 bytes: %d", n)
+	}
+}
+
+func TestUint64TableProbeOneByOne(t *testing.T) {
+	const N = 500
+	table := NewUint64Table(0, 0.9)
+
+	for n := 0; n < 2; n++ {
+		// Do two passes, both should behave the same.
+		for i := 1; i <= N; i++ {
+			k := [1]uint64{}
+			v := [1]int32{}
+
+			k[0] = uint64(i)
+			table.Probe(k[:], v[:])
+
+			if v[0] != int32(i-1) {
+				t.Errorf("wrong value probed for key=%d: want=%d got=%d", i, i-1, v[0])
+			}
+		}
+	}
+}
+
+func TestUint64TableProbeBulk(t *testing.T) {
+	const N = 999
+	table := NewUint64Table(0, 0.9)
+
+	k := make([]uint64, N)
+	v := make([]int32, N)
+
+	for i := range k {
+		k[i] = uint64(i)
+	}
+
+	for n := 0; n < 2; n++ {
+		table.Probe(k, v)
+
+		for i := range v {
+			if v[i] != int32(i) {
+				t.Errorf("wrong value probed for key=%d: want=%d got=%d", k[i], i, v[i])
+			}
+		}
+
+		if t.Failed() {
+			break
+		}
+
+		for i := range v {
+			v[i] = 0
+		}
+	}
+}
+
+func TestUint128TableProbeOneByOne(t *testing.T) {
+	const N = 500
+	table := NewUint128Table(0, 0.9)
+
+	for n := 0; n < 2; n++ {
+		// Do two passes, both should behave the same.
+		for i := 1; i <= N; i++ {
+			k := [1][16]byte{}
+			v := [1]int32{}
+
+			binary.LittleEndian.PutUint64(k[0][:8], uint64(i))
+			table.Probe(k[:], v[:])
+
+			if v[0] != int32(i-1) {
+				t.Errorf("wrong value probed for key=%x: want=%d got=%d", i, i-1, v[0])
+			}
+		}
+	}
+}
+
+func TestUint128TableProbeBulk(t *testing.T) {
+	const N = 999
+	table := NewUint128Table(0, 0.9)
+
+	k := make([][16]byte, N)
+	v := make([]int32, N)
+
+	for i := range k {
+		binary.LittleEndian.PutUint64(k[i][:8], uint64(i))
+	}
+
+	for n := 0; n < 2; n++ {
+		table.Probe(k, v)
+
+		for i := range v {
+			if v[i] != int32(i) {
+				t.Errorf("wrong value probed for key=%x: want=%d got=%d", k[i], i, v[i])
+			}
+		}
+
+		if t.Failed() {
+			break
+		}
+
+		for i := range v {
+			v[i] = 0
+		}
+	}
+}
+
+const (
+	benchmarkProbesPerLoop = 500
+	benchmarkMaxLoad       = 0.9
+)
+
+type uint32Table interface {
+	Reset()
+	Len() int
+	Probe([]uint32, []int32) int
+}
+
+type uint32Map map[uint32]int32
+
+func (m uint32Map) Reset() {
+	for k := range m {
+		delete(m, k)
+	}
+}
+
+func (m uint32Map) Len() int {
+	return len(m)
+}
+
+func (m uint32Map) Probe(keys []uint32, values []int32) (n int) {
+	_ = values[:len(keys)]
+
+	for i, k := range keys {
+		v, ok := m[k]
+		if !ok {
+			v = int32(len(m))
+			m[k] = v
+			n++
+		}
+		values[i] = v
+	}
+
+	return n
+}
+
+func BenchmarkUint32Table(b *testing.B) {
+	benchmarkUint32Table(b, func(size int) uint32Table { return NewUint32Table(size, benchmarkMaxLoad) })
+}
+
+func BenchmarkGoUint32Map(b *testing.B) {
+	benchmarkUint32Table(b, func(size int) uint32Table { return make(uint32Map, size) })
+}
+
+func benchmarkUint32Table(b *testing.B, newTable func(size int) uint32Table) {
+	for n := 100; n <= 1e6; n *= 10 {
+		table := newTable(0)
+		keys, values := generateUint32Table(n)
+
+		b.Run(fmt.Sprintf("N=%d", n), func(b *testing.B) {
+			benchmarkUint32Loop(b, table.Probe, keys, values)
+		})
+	}
+}
+
+func benchmarkUint32Loop(b *testing.B, f func([]uint32, []int32) int, keys []uint32, values []int32) {
+	i := 0
+	j := benchmarkProbesPerLoop
+	b.SetBytes(4 * int64(benchmarkProbesPerLoop))
+
+	_ = keys[:len(values)]
+	_ = values[:len(keys)]
+	start := time.Now()
+
+	for k := 0; k < b.N; k++ {
+		if j > len(keys) {
+			j = len(keys)
+		}
+		f(keys[i:j:j], values[i:j:j])
+		if j == len(keys) {
+			i, j = 0, benchmarkProbesPerLoop
+		} else {
+			i, j = j, j+benchmarkProbesPerLoop
+		}
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(benchmarkProbesPerLoop*b.N)/seconds, "probe/s")
+}
+
+func generateUint32Table(n int) ([]uint32, []int32) {
+	prng := rand.New(rand.NewSource(int64(n)))
+	keys := make([]uint32, n)
+	values := make([]int32, n)
+
+	for i := range keys {
+		keys[i] = prng.Uint32()
+	}
+
+	return keys, values
+}
+
+type uint64Table interface {
+	Reset()
+	Len() int
+	Probe([]uint64, []int32) int
+}
+
+type uint64Map map[uint64]int32
+
+func (m uint64Map) Reset() {
+	for k := range m {
+		delete(m, k)
+	}
+}
+
+func (m uint64Map) Len() int {
+	return len(m)
+}
+
+func (m uint64Map) Probe(keys []uint64, values []int32) (n int) {
+	_ = values[:len(keys)]
+
+	for i, k := range keys {
+		v, ok := m[k]
+		if !ok {
+			v = int32(len(m))
+			m[k] = v
+			n++
+		}
+		values[i] = v
+	}
+
+	return n
+}
+
+func BenchmarkUint64Table(b *testing.B) {
+	benchmarkUint64Table(b, func(size int) uint64Table { return NewUint64Table(size, benchmarkMaxLoad) })
+}
+
+func BenchmarkGoUint64Map(b *testing.B) {
+	benchmarkUint64Table(b, func(size int) uint64Table { return make(uint64Map, size) })
+}
+
+func benchmarkUint64Table(b *testing.B, newTable func(size int) uint64Table) {
+	for n := 100; n <= 1e6; n *= 10 {
+		table := newTable(0)
+		keys, values := generateUint64Table(n)
+
+		b.Run(fmt.Sprintf("N=%d", n), func(b *testing.B) {
+			benchmarkUint64Loop(b, table.Probe, keys, values)
+		})
+	}
+}
+
+func benchmarkUint64Loop(b *testing.B, f func([]uint64, []int32) int, keys []uint64, values []int32) {
+	i := 0
+	j := benchmarkProbesPerLoop
+	b.SetBytes(8 * int64(benchmarkProbesPerLoop))
+
+	_ = keys[:len(values)]
+	_ = values[:len(keys)]
+	start := time.Now()
+
+	for k := 0; k < b.N; k++ {
+		if j > len(keys) {
+			j = len(keys)
+		}
+		f(keys[i:j:j], values[i:j:j])
+		if j == len(keys) {
+			i, j = 0, benchmarkProbesPerLoop
+		} else {
+			i, j = j, j+benchmarkProbesPerLoop
+		}
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(benchmarkProbesPerLoop*b.N)/seconds, "probe/s")
+}
+
+func generateUint64Table(n int) ([]uint64, []int32) {
+	prng := rand.New(rand.NewSource(int64(n)))
+	keys := make([]uint64, n)
+	values := make([]int32, n)
+
+	for i := range keys {
+		keys[i] = prng.Uint64()
+	}
+
+	return keys, values
+}
+
+type uint128Table interface {
+	Reset()
+	Len() int
+	Probe([][16]byte, []int32) int
+}
+
+type uint128Map map[[16]byte]int32
+
+func (m uint128Map) Reset() {
+	for k := range m {
+		delete(m, k)
+	}
+}
+
+func (m uint128Map) Len() int {
+	return len(m)
+}
+
+func (m uint128Map) Probe(keys [][16]byte, values []int32) (n int) {
+	_ = values[:len(keys)]
+
+	for i, k := range keys {
+		v, ok := m[k]
+		if !ok {
+			v = int32(len(m))
+			m[k] = v
+			n++
+		}
+		values[i] = v
+	}
+
+	return n
+}
+
+func BenchmarkUint128Table(b *testing.B) {
+	benchmarkUint128Table(b, func(size int) uint128Table { return NewUint128Table(size, benchmarkMaxLoad) })
+}
+
+func BenchmarkGoUint128Map(b *testing.B) {
+	benchmarkUint128Table(b, func(size int) uint128Table { return make(uint128Map, size) })
+}
+
+func benchmarkUint128Table(b *testing.B, newTable func(size int) uint128Table) {
+	for n := 100; n <= 1e6; n *= 10 {
+		table := newTable(0)
+		keys, values := generateUint128Table(n)
+
+		b.Run(fmt.Sprintf("N=%d", n), func(b *testing.B) {
+			benchmarkUint128Loop(b, table.Probe, keys, values)
+		})
+	}
+}
+
+func benchmarkUint128Loop(b *testing.B, f func([][16]byte, []int32) int, keys [][16]byte, values []int32) {
+	i := 0
+	j := benchmarkProbesPerLoop
+	b.SetBytes(16 * int64(benchmarkProbesPerLoop))
+
+	_ = keys[:len(values)]
+	_ = values[:len(keys)]
+	start := time.Now()
+
+	for k := 0; k < b.N; k++ {
+		if j > len(keys) {
+			j = len(keys)
+		}
+		f(keys[i:j:j], values[i:j:j])
+		if j == len(keys) {
+			i, j = 0, benchmarkProbesPerLoop
+		} else {
+			i, j = j, j+benchmarkProbesPerLoop
+		}
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(benchmarkProbesPerLoop*b.N)/seconds, "probe/s")
+}
+
+func generateUint128Table(n int) ([][16]byte, []int32) {
+	prng := rand.New(rand.NewSource(int64(n)))
+	keys := make([][16]byte, n)
+	values := make([]int32, n)
+
+	for i := range keys {
+		prng.Read(keys[i][:])
+	}
+
+	return keys, values
+}
diff --git a/hashprobe/wyhash/wyhash_test.go b/hashprobe/wyhash/wyhash_test.go
new file mode 100644
index 0000000..07c3dcf
--- /dev/null
+++ b/hashprobe/wyhash/wyhash_test.go
@@ -0,0 +1,162 @@
+package wyhash
+
+import (
+	"encoding/binary"
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func TestHash32(t *testing.T) {
+	if h := Hash32(42, 1); h != 0xda93b6f668a0496e {
+		t.Errorf("hash mismatch: %08x", h)
+	}
+}
+
+func TestMultiHash32(t *testing.T) {
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N]uint32{}
+	seed := uintptr(32)
+
+	for i := range values {
+		values[i] = uint32(i)
+	}
+
+	MultiHash32(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash32(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%08x got=%08x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func BenchmarkHash32(b *testing.B) {
+	b.SetBytes(8)
+	value := rand.Uint32()
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		value = uint32(Hash32(value, seed))
+		return 1
+	})
+}
+
+func BenchmarkMultiHash32(b *testing.B) {
+	hashes := [512]uintptr{}
+	values := [512]uint32{}
+	b.SetBytes(4 * int64(len(hashes)))
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		MultiHash32(hashes[:], values[:], seed)
+		return len(hashes)
+	})
+}
+
+func TestHash64(t *testing.T) {
+	if h := Hash64(42, 1); h != 0x6e69a6ede6b5a25e {
+		t.Errorf("hash mismatch: %016x", h)
+	}
+}
+
+func TestMultiHash64(t *testing.T) {
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N]uint64{}
+	seed := uintptr(64)
+
+	for i := range values {
+		values[i] = uint64(i)
+	}
+
+	MultiHash64(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash64(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%016x got=%016x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func BenchmarkHash64(b *testing.B) {
+	b.SetBytes(8)
+	value := rand.Uint64()
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		value = uint64(Hash64(value, seed))
+		return 1
+	})
+}
+
+func BenchmarkMultiHash64(b *testing.B) {
+	hashes := [512]uintptr{}
+	values := [512]uint64{}
+	b.SetBytes(8 * int64(len(hashes)))
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		MultiHash64(hashes[:], values[:], seed)
+		return len(hashes)
+	})
+}
+
+func TestHash128(t *testing.T) {
+	if h := Hash128([16]byte{0: 42}, 1); h != 0xcd09fcdae9a79e7c {
+		t.Errorf("hash mismatch: %016x", h)
+	}
+}
+
+func TestMultiHash128(t *testing.T) {
+	const N = 10
+	hashes := [N]uintptr{}
+	values := [N][16]byte{}
+	seed := uintptr(64)
+
+	for i := range values {
+		binary.LittleEndian.PutUint64(values[i][:8], uint64(i))
+	}
+
+	MultiHash128(hashes[:], values[:], seed)
+
+	for i := range values {
+		h := Hash128(values[i], seed)
+
+		if h != hashes[i] {
+			t.Errorf("hash(%d): want=%016x got=%016x", values[i], h, hashes[i])
+		}
+	}
+}
+
+func BenchmarkHash128(b *testing.B) {
+	b.SetBytes(8)
+	hash := uintptr(0)
+	value := [16]byte{}
+	binary.LittleEndian.PutUint64(value[:8], rand.Uint64())
+	binary.LittleEndian.PutUint64(value[8:], rand.Uint64())
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		hash = Hash128(value, seed)
+		return 1
+	})
+	_ = hash
+}
+
+func BenchmarkMultiHash128(b *testing.B) {
+	hashes := [512]uintptr{}
+	values := [512][16]byte{}
+	b.SetBytes(16 * int64(len(hashes)))
+	benchmarkHashThroughput(b, func(seed uintptr) int {
+		MultiHash128(hashes[:], values[:], seed)
+		return len(hashes)
+	})
+}
+
+func benchmarkHashThroughput(b *testing.B, f func(seed uintptr) int) {
+	hashes := int64(0)
+	start := time.Now()
+
+	for i := 0; i < b.N; i++ {
+		hashes += int64(f(uintptr(i)))
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(hashes)/seconds, "hash/s")
+}
diff --git a/null_test.go b/null_test.go
index 4a31ec0..46b3a66 100644
--- a/null_test.go
+++ b/null_test.go
@@ -5,7 +5,6 @@ package parquet
 import (
 	"reflect"
 	"testing"
-	"unsafe"
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/internal/quick"
@@ -37,7 +36,6 @@ func testNullIndex[T comparable](t *testing.T) {
 				return true
 			}
 
-			size := unsafe.Sizeof(zero)
 			want := make([]uint64, (len(data)+63)/64)
 			got := make([]uint64, (len(data)+63)/64)
 
@@ -48,8 +46,8 @@ func testNullIndex[T comparable](t *testing.T) {
 			}
 
 			array := makeArrayOf(data)
-			nullIndex[T](want, array, size, 0)
-			nullIndexFuncOf(reflect.TypeOf(zero))(got, array, size, 0)
+			nullIndex[T](want, array)
+			nullIndexFuncOf(reflect.TypeOf(zero))(got, array)
 
 			if !reflect.DeepEqual(want, got) {
 				t.Errorf("unexpected null index\nwant = %064b\ngot  = %064b", want, got)
@@ -86,15 +84,14 @@ func benchmarkNullIndex[T any](b *testing.B) {
 
 	var zero T
 	typ := reflect.TypeOf(zero)
-	size := unsafe.Sizeof(zero)
 	null := nullIndexFuncOf(typ)
 	data := makeArrayOf(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)
+			null(bits, data)
 		}
-		b.SetBytes(int64(size * N))
+		b.SetBytes(int64(typ.Size() * N))
 	})
 }
diff --git a/sparse/gather_test.go b/sparse/gather_test.go
new file mode 100644
index 0000000..7ed1ed7
--- /dev/null
+++ b/sparse/gather_test.go
@@ -0,0 +1,236 @@
+package sparse_test
+
+import (
+	"encoding/binary"
+	"fmt"
+	"math"
+	"testing"
+	"time"
+	"unsafe"
+
+	"github.com/segmentio/parquet-go/sparse"
+)
+
+const (
+	benchmarkGatherPerLoop = 1000
+)
+
+func ExampleGatherUint32() {
+	type point2D struct{ X, Y uint32 }
+
+	buf := make([]point2D, 10)
+	dst := make([]uint32, 10)
+	src := sparse.UnsafeUint32Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		buf[i].X = math.MaxUint32
+		buf[i].Y = uint32(2 * i)
+	}
+
+	n := sparse.GatherUint32(dst, src)
+
+	for i, v := range dst[:n] {
+		fmt.Printf("points[%d].Y = %d\n", i, v)
+	}
+
+	// Output:
+	// points[0].Y = 0
+	// points[1].Y = 2
+	// points[2].Y = 4
+	// points[3].Y = 6
+	// points[4].Y = 8
+	// points[5].Y = 10
+	// points[6].Y = 12
+	// points[7].Y = 14
+	// points[8].Y = 16
+	// points[9].Y = 18
+}
+
+func ExampleGatherUint64() {
+	type point2D struct{ X, Y uint64 }
+
+	buf := make([]point2D, 10)
+	dst := make([]uint64, 10)
+	src := sparse.UnsafeUint64Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		buf[i].X = math.MaxUint64
+		buf[i].Y = uint64(2 * i)
+	}
+
+	n := sparse.GatherUint64(dst, src)
+
+	for i, v := range dst[:n] {
+		fmt.Printf("points[%d].Y = %v\n", i, v)
+	}
+
+	// Output:
+	// points[0].Y = 0
+	// points[1].Y = 2
+	// points[2].Y = 4
+	// points[3].Y = 6
+	// points[4].Y = 8
+	// points[5].Y = 10
+	// points[6].Y = 12
+	// points[7].Y = 14
+	// points[8].Y = 16
+	// points[9].Y = 18
+}
+
+func ExampleGatherUint128() {
+	type point2D struct{ X, Y [16]byte }
+
+	buf := make([]point2D, 10)
+	dst := make([][16]byte, 10)
+	src := sparse.UnsafeUint128Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		x := uint64(math.MaxUint64)
+		y := uint64(2 * i)
+		binary.LittleEndian.PutUint64(buf[i].X[:], x)
+		binary.LittleEndian.PutUint64(buf[i].Y[:], y)
+	}
+
+	n := sparse.GatherUint128(dst, src)
+
+	for i, v := range dst[:n] {
+		fmt.Printf("points[%d].Y = %v\n", i, binary.LittleEndian.Uint64(v[:]))
+	}
+
+	// Output:
+	// points[0].Y = 0
+	// points[1].Y = 2
+	// points[2].Y = 4
+	// points[3].Y = 6
+	// points[4].Y = 8
+	// points[5].Y = 10
+	// points[6].Y = 12
+	// points[7].Y = 14
+	// points[8].Y = 16
+	// points[9].Y = 18
+}
+
+func TestGatherUint32(t *testing.T) {
+	type point2D struct{ X, Y uint32 }
+
+	const N = 100
+	buf := make([]point2D, N+1)
+	dst := make([]uint32, N)
+	src := sparse.UnsafeUint32Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		buf[i].X = math.MaxUint32
+		buf[i].Y = uint32(2 * i)
+	}
+
+	if n := sparse.GatherUint32(dst, src); n != N {
+		t.Errorf("wrong number of values gathered: want=%d got=%d", N, n)
+	}
+
+	for i, v := range dst {
+		if v != uint32(2*i) {
+			t.Errorf("wrong value gathered at index %d: want=%d got=%d", i, 2*i, v)
+		}
+	}
+}
+
+func TestGatherUint64(t *testing.T) {
+	type point2D struct{ X, Y uint64 }
+
+	const N = 100
+	buf := make([]point2D, N+1)
+	dst := make([]uint64, N)
+	src := sparse.UnsafeUint64Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		buf[i].X = math.MaxUint64
+		buf[i].Y = uint64(2 * i)
+	}
+
+	if n := sparse.GatherUint64(dst, src); n != N {
+		t.Errorf("wrong number of values gathered: want=%d got=%d", N, n)
+	}
+
+	for i, v := range dst {
+		if v != uint64(2*i) {
+			t.Errorf("wrong value gathered at index %d: want=%d got=%d", i, 2*i, v)
+		}
+	}
+}
+
+func TestGatherUint128(t *testing.T) {
+	type point2D struct{ X, Y [16]byte }
+
+	const N = 100
+	buf := make([]point2D, N+1)
+	dst := make([][16]byte, N)
+	src := sparse.UnsafeUint128Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	for i := range buf {
+		x := uint64(math.MaxUint64)
+		y := uint64(2 * i)
+		binary.LittleEndian.PutUint64(buf[i].X[:], x)
+		binary.LittleEndian.PutUint64(buf[i].Y[:], y)
+	}
+
+	if n := sparse.GatherUint128(dst, src); n != N {
+		t.Errorf("wrong number of values gathered: want=%d got=%d", N, n)
+	}
+
+	for i, v := range dst {
+		if y := binary.LittleEndian.Uint64(v[:]); y != uint64(2*i) {
+			t.Errorf("wrong value gathered at index %d: want=%d got=%d", i, 2*i, y)
+		}
+	}
+}
+
+func BenchmarkGather32(b *testing.B) {
+	type point2D struct{ X, Y uint32 }
+
+	buf := make([]point2D, benchmarkGatherPerLoop)
+	dst := make([]uint32, benchmarkGatherPerLoop)
+	src := sparse.UnsafeUint32Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	b.SetBytes(4 * benchmarkGatherPerLoop)
+	benchmarkThroughput(b, "gather", func() int {
+		return sparse.GatherUint32(dst, src)
+	})
+}
+
+func BenchmarkGather64(b *testing.B) {
+	type point2D struct{ X, Y uint64 }
+
+	buf := make([]point2D, benchmarkGatherPerLoop)
+	dst := make([]uint64, benchmarkGatherPerLoop)
+	src := sparse.UnsafeUint64Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	b.SetBytes(8 * benchmarkGatherPerLoop)
+	benchmarkThroughput(b, "gather", func() int {
+		return sparse.GatherUint64(dst, src)
+	})
+}
+
+func BenchmarkGather128(b *testing.B) {
+	type point2D struct{ X, Y [16]byte }
+
+	buf := make([]point2D, benchmarkGatherPerLoop)
+	dst := make([][16]byte, benchmarkGatherPerLoop)
+	src := sparse.UnsafeUint128Array(unsafe.Pointer(&buf[0].Y), len(buf), unsafe.Sizeof(buf[0]))
+
+	b.SetBytes(16 * benchmarkGatherPerLoop)
+	benchmarkThroughput(b, "gather", func() int {
+		return sparse.GatherUint128(dst, src)
+	})
+}
+
+func benchmarkThroughput(b *testing.B, m string, f func() int) {
+	start := time.Now()
+	count := 0
+
+	for i := 0; i < b.N; i++ {
+		count += f()
+	}
+
+	seconds := time.Since(start).Seconds()
+	b.ReportMetric(float64(count)/seconds, m+"/s")
+}