Repository: segmentio/parquet-go
Pull request #187 (2022-05-21 19:31:02)

optimize and simplify DELTA_BYTE_ARRAY encoding
While working on #186, I noticed a simple optimization that could be made to the DELTA_BYTE_ARRAY encoding. CPU profiles showed using the intermediary `bytes.Buffer` to construct the values made of the last value's prefix and current value's suffix had a measurable overhead.

![image](https://user-images.githubusercontent.com/865510/169666512-c503fdae-cfde-42ab-b246-5e1f4d411984.png)

Since we are working with byte slices on the input and output, this intermediary value is actually unnecessary, we can use a byte slice in the destination buffer where we wrote the last value instead. This change in the implementation also means that we do not need the optimization to maintain an internal pool of `bytes.Buffer`, since we are not using the intermediary buffer we can remove this complexity entirely.

```
name                                old time/op   new time/op   delta
Decode/DELTA_BYTE_ARRAY/byte_array    309µs ± 0%    279µs ± 0%   -9.68%  (p=0.000 n=10+10)

name                                old speed     new speed     delta
Decode/DELTA_BYTE_ARRAY/byte_array  468MB/s ± 0%  518MB/s ± 0%  +10.71%  (p=0.000 n=10+10)

```
```
name                                             old time/op   new time/op   delta
Decode/DELTA_BYTE_ARRAY/fixed_length_byte_array    194µs ± 0%    177µs ± 0%  -8.64%  (p=0.000 n=9+9)

name                                             old speed     new speed     delta
Decode/DELTA_BYTE_ARRAY/fixed_length_byte_array  824MB/s ± 0%  901MB/s ± 0%  +9.46%  (p=0.000 n=9+9)
```


Patch:
diff --git a/bloom/block_amd64.go b/bloom/block_amd64.go
index 16fb17f..1113010 100644
--- a/bloom/block_amd64.go
+++ b/bloom/block_amd64.go
@@ -2,6 +2,8 @@
 
 package bloom
 
+import "golang.org/x/sys/cpu"
+
 // The functions in this file are SIMD-optimized versions of the functions
 // declared in block_optimized.go for x86 targets.
 //
@@ -24,6 +26,7 @@ package bloom
 // block_optimized.go; the delta comparing to functions in block_default.go is
 // significantly larger but not very interesting since those functions have no
 // practical use cases.
+var hasAVX2 = cpu.X86.HasAVX2
 
 //go:noescape
 func blockInsert(b *Block, x uint32)
diff --git a/bloom/block_amd64.s b/bloom/block_amd64.s
index b8ade61..713aea2 100644
--- a/bloom/block_amd64.s
+++ b/bloom/block_amd64.s
@@ -2,6 +2,15 @@
 
 #include "textflag.h"
 
+#define salt0 0x47b6137b
+#define salt1 0x44974d91
+#define salt2 0x8824ad5b
+#define salt3 0xa2b7289d
+#define salt4 0x705495c7
+#define salt5 0x2df1424b
+#define salt6 0x9efc4947
+#define salt7 0x5c6bfb31
+
 DATA ones+0(SB)/4, $1
 DATA ones+4(SB)/4, $1
 DATA ones+8(SB)/4, $1
@@ -12,14 +21,14 @@ DATA ones+24(SB)/4, $1
 DATA ones+28(SB)/4, $1
 GLOBL ones(SB), RODATA|NOPTR, $32
 
-DATA salt+0(SB)/4, $0x47b6137b
-DATA salt+4(SB)/4, $0x44974d91
-DATA salt+8(SB)/4, $0x8824ad5b
-DATA salt+12(SB)/4, $0xa2b7289d
-DATA salt+16(SB)/4, $0x705495c7
-DATA salt+20(SB)/4, $0x2df1424b
-DATA salt+24(SB)/4, $0x9efc4947
-DATA salt+28(SB)/4, $0x5c6bfb31
+DATA salt+0(SB)/4, $salt0
+DATA salt+4(SB)/4, $salt1
+DATA salt+8(SB)/4, $salt2
+DATA salt+12(SB)/4, $salt3
+DATA salt+16(SB)/4, $salt4
+DATA salt+20(SB)/4, $salt5
+DATA salt+24(SB)/4, $salt6
+DATA salt+28(SB)/4, $salt7
 GLOBL salt(SB), RODATA|NOPTR, $32
 
 // This initial block is a SIMD implementation of the mask function declared in
@@ -46,19 +55,52 @@ GLOBL salt(SB), RODATA|NOPTR, $32
     VPSRLD $27, tmp, tmp \
     VPSLLVD tmp, dst, dst
 
+#define insert(salt, src, dst) \
+    MOVL src, CX \
+    IMULL salt, CX \
+    SHRL $27, CX \
+    MOVL $1, DX \
+    SHLL CX, DX \
+    ORL DX, dst
+
+#define check(salt, b, x) \
+    MOVL b, CX \
+    MOVL x, DX \
+    IMULL salt, DX \
+    SHRL $27, DX \
+    BTL DX, CX \
+    JAE notfound
+
 // func blockInsert(b *Block, x uint32)
 TEXT ·blockInsert(SB), NOSPLIT, $0-16
     MOVQ b+0(FP), AX
+    CMPB ·hasAVX2(SB), $0
+    JE fallback
+avx2:
     generateMask(x+8(FP), Y1, Y0)
     // Set all 1 bits of the mask in the bloom filter block.
     VPOR (AX), Y0, Y0
     VMOVDQU Y0, (AX)
     VZEROUPPER
     RET
+fallback:
+    MOVL x+8(FP), BX
+    insert($salt0, BX, 0(AX))
+    insert($salt1, BX, 4(AX))
+    insert($salt2, BX, 8(AX))
+    insert($salt3, BX, 12(AX))
+    insert($salt4, BX, 16(AX))
+    insert($salt5, BX, 20(AX))
+    insert($salt6, BX, 24(AX))
+    insert($salt7, BX, 28(AX))
+    RET
 
 // func blockCheck(b *Block, x uint32) bool
 TEXT ·blockCheck(SB), NOSPLIT, $0-17
     MOVQ b+0(FP), AX
+    CMPB ·hasAVX2(SB), $0
+    JE fallback
+avx2:
     generateMask(x+8(FP), Y1, Y0)
     // Compare the 1 bits of the mask with the bloom filter block, then compare
     // the result with the mask, expecting equality if the value `x` was present
@@ -68,3 +110,20 @@ TEXT ·blockCheck(SB), NOSPLIT, $0-17
     SETCS ret+16(FP)   // return CF == 1
     VZEROUPPER
     RET
+fallback:
+    MOVL x+8(FP), BX
+    check($salt0, 0(AX), BX)
+    check($salt1, 4(AX), BX)
+    check($salt2, 8(AX), BX)
+    check($salt3, 12(AX), BX)
+    check($salt4, 16(AX), BX)
+    check($salt5, 20(AX), BX)
+    check($salt6, 24(AX), BX)
+    check($salt7, 28(AX), BX)
+    MOVB $1, CX
+    JMP done
+notfound:
+    XORB CX, CX
+done:
+    MOVB CX, ret+16(FP)
+    RET
diff --git a/bloom/block_optimized.go b/bloom/block_optimized.go
index e879d15..d67bec1 100644
--- a/bloom/block_optimized.go
+++ b/bloom/block_optimized.go
@@ -24,41 +24,26 @@ package bloom
 // The benchmarks measure throughput based on the byte size of a bloom filter
 // block.
 
-func mask(x uint32) Block {
-	return Block{
-		0: 1 << ((x * salt0) >> 27),
-		1: 1 << ((x * salt1) >> 27),
-		2: 1 << ((x * salt2) >> 27),
-		3: 1 << ((x * salt3) >> 27),
-		4: 1 << ((x * salt4) >> 27),
-		5: 1 << ((x * salt5) >> 27),
-		6: 1 << ((x * salt6) >> 27),
-		7: 1 << ((x * salt7) >> 27),
-	}
-}
-
 func (b *Block) Insert(x uint32) {
-	masked := mask(x)
-	b[0] |= masked[0]
-	b[1] |= masked[1]
-	b[2] |= masked[2]
-	b[3] |= masked[3]
-	b[4] |= masked[4]
-	b[5] |= masked[5]
-	b[6] |= masked[6]
-	b[7] |= masked[7]
+	b[0] |= 1 << ((x * salt0) >> 27)
+	b[1] |= 1 << ((x * salt1) >> 27)
+	b[2] |= 1 << ((x * salt2) >> 27)
+	b[3] |= 1 << ((x * salt3) >> 27)
+	b[4] |= 1 << ((x * salt4) >> 27)
+	b[5] |= 1 << ((x * salt5) >> 27)
+	b[6] |= 1 << ((x * salt6) >> 27)
+	b[7] |= 1 << ((x * salt7) >> 27)
 }
 
 func (b *Block) Check(x uint32) bool {
-	masked := mask(x)
-	return ((b[0] & masked[0]) != 0) &&
-		((b[1] & masked[1]) != 0) &&
-		((b[2] & masked[2]) != 0) &&
-		((b[3] & masked[3]) != 0) &&
-		((b[4] & masked[4]) != 0) &&
-		((b[5] & masked[5]) != 0) &&
-		((b[6] & masked[6]) != 0) &&
-		((b[7] & masked[7]) != 0)
+	return ((b[0] & (1 << ((x * salt0) >> 27))) != 0) &&
+		((b[1] & (1 << ((x * salt1) >> 27))) != 0) &&
+		((b[2] & (1 << ((x * salt2) >> 27))) != 0) &&
+		((b[3] & (1 << ((x * salt3) >> 27))) != 0) &&
+		((b[4] & (1 << ((x * salt4) >> 27))) != 0) &&
+		((b[5] & (1 << ((x * salt5) >> 27))) != 0) &&
+		((b[6] & (1 << ((x * salt6) >> 27))) != 0) &&
+		((b[7] & (1 << ((x * salt7) >> 27))) != 0)
 }
 
 func (f SplitBlockFilter) insertBulk(x []uint64) {
diff --git a/bloom/filter_amd64.s b/bloom/filter_amd64.s
index e216210..cfa75e6 100644
--- a/bloom/filter_amd64.s
+++ b/bloom/filter_amd64.s
@@ -2,6 +2,15 @@
 
 #include "textflag.h"
 
+#define salt0 0x47b6137b
+#define salt1 0x44974d91
+#define salt2 0x8824ad5b
+#define salt3 0xa2b7289d
+#define salt4 0x705495c7
+#define salt5 0x2df1424b
+#define salt6 0x9efc4947
+#define salt7 0x5c6bfb31
+
 // See block_amd64.s for a description of this algorithm.
 #define generateMask(src, dst) \
     VMOVDQA ones(SB), dst \
@@ -32,23 +41,41 @@
     MOVQ tmpXMM, r2 \
     VPEXTRQ $1, tmpXMM, r3
 
+#define insert(salt, src, dst) \
+    MOVL src, CX \
+    IMULL salt, CX \
+    SHRL $27, CX \
+    MOVL $1, DX \
+    SHLL CX, DX \
+    ORL DX, dst
+
+#define check(salt, b, x) \
+    MOVL b, CX \
+    MOVL x, DX \
+    IMULL salt, DX \
+    SHRL $27, DX \
+    BTL DX, CX \
+    JAE notfound
+
 // func filterInsertBulk(f []Block, x []uint64)
 TEXT ·filterInsertBulk(SB), NOSPLIT, $0-48
     MOVQ f_base+0(FP), AX
     MOVQ f_len+8(FP), CX
     MOVQ x_base+24(FP), BX
     MOVQ x_len+32(FP), DX
+    CMPB ·hasAVX2(SB), $0
+    JE fallback
+avx2:
     VPBROADCASTQ f_base+8(FP), Y0
-
     // Loop initialization, SI holds the current index in `x`, DI is the number
     // of elements in `x` rounded down to the nearest multiple of 4.
     XORQ SI, SI
     MOVQ DX, DI
     SHRQ $2, DI
     SHLQ $2, DI
-loop4x64:
+avx2loop4x64:
     CMPQ SI, DI
-    JAE loop
+    JAE avx2loop1x64
 
     // The masks and indexes for 4 input hashes are computed in each loop
     // iteration. The hashes are loaded in Y1 so we can use vector instructions
@@ -84,22 +111,44 @@ loop4x64:
     applyMask(Y9, (AX)(R11*1))
 
     ADDQ $4, SI
-    JMP loop4x64
-loop:
+    JMP avx2loop4x64
+avx2loop1x64:
     // Compute trailing elements in `x` if the length was not a multiple of 4.
     // This is the same algorithm as the one in the loop4x64 section, working
     // on a single mask/block pair at a time.
     CMPQ SI, DX
-    JE done
+    JE avx2done
     MOVQ (BX)(SI*8), R8
     VPBROADCASTD (BX)(SI*8), Y0
     fasthash1x64(CX, R8)
     generateMask(Y0, Y1)
     applyMask(Y1, (AX)(R8*1))
     INCQ SI
+    JMP avx2loop1x64
+avx2done:
+    VZEROUPPER
+    JMP done
+fallback:
+    XORQ SI, SI
+    MOVQ DX, DI
+    MOVQ CX, R10
+loop:
+    CMPQ SI, DI
+    JE done
+    MOVLQZX (BX)(SI*8), R8
+    MOVQ (BX)(SI*8), R9
+    fasthash1x64(R10, R9)
+    insert($salt0, R8, 0(AX)(R9*1))
+    insert($salt1, R8, 4(AX)(R9*1))
+    insert($salt2, R8, 8(AX)(R9*1))
+    insert($salt3, R8, 12(AX)(R9*1))
+    insert($salt4, R8, 16(AX)(R9*1))
+    insert($salt5, R8, 20(AX)(R9*1))
+    insert($salt6, R8, 24(AX)(R9*1))
+    insert($salt7, R8, 28(AX)(R9*1))
+    INCQ SI
     JMP loop
 done:
-    VZEROUPPER
     RET
 
 // func filterInsert(f []Block, x uint64)
@@ -107,23 +156,59 @@ TEXT ·filterInsert(SB), NOSPLIT, $0-32
     MOVQ f_base+0(FP), AX
     MOVQ f_len+8(FP), BX
     MOVQ x+24(FP), CX
-    VPBROADCASTD x+24(FP), Y1
     fasthash1x64(BX, CX)
+    CMPB ·hasAVX2(SB), $0
+    JE fallback
+avx2:
+    VPBROADCASTD x+24(FP), Y1
     generateMask(Y1, Y0)
     applyMask(Y0, (AX)(CX*1))
     VZEROUPPER
     RET
+fallback:
+    ADDQ CX, AX
+    MOVL x+24(FP), BX
+    insert($salt0, BX, 0(AX))
+    insert($salt1, BX, 4(AX))
+    insert($salt2, BX, 8(AX))
+    insert($salt3, BX, 12(AX))
+    insert($salt4, BX, 16(AX))
+    insert($salt5, BX, 20(AX))
+    insert($salt6, BX, 24(AX))
+    insert($salt7, BX, 28(AX))
+    RET
 
 // func filterCheck(f []Block, x uint64) bool
 TEXT ·filterCheck(SB), NOSPLIT, $0-33
     MOVQ f_base+0(FP), AX
     MOVQ f_len+8(FP), BX
     MOVQ x+24(FP), CX
-    VPBROADCASTD x+24(FP), Y1
     fasthash1x64(BX, CX)
+    CMPB ·hasAVX2(SB), $0
+    JE fallback
+avx2:
+    VPBROADCASTD x+24(FP), Y1
     generateMask(Y1, Y0)
     VPAND (AX)(CX*1), Y0, Y1
     VPTEST Y0, Y1
     SETCS ret+32(FP)
     VZEROUPPER
     RET
+fallback:
+    ADDQ CX, AX
+    MOVL x+24(FP), BX
+    check($salt0, 0(AX), BX)
+    check($salt1, 4(AX), BX)
+    check($salt2, 8(AX), BX)
+    check($salt3, 12(AX), BX)
+    check($salt4, 16(AX), BX)
+    check($salt5, 20(AX), BX)
+    check($salt6, 24(AX), BX)
+    check($salt7, 28(AX), BX)
+    MOVB $1, CX
+    JMP done
+notfound:
+    XORB CX, CX
+done:
+    MOVB CX, ret+32(FP)
+    RET
diff --git a/encoding/delta/byte_array.go b/encoding/delta/byte_array.go
index 2d35562..83ec707 100644
--- a/encoding/delta/byte_array.go
+++ b/encoding/delta/byte_array.go
@@ -101,9 +101,15 @@ func (e *ByteArrayEncoding) encode(dst []byte, numValues int, valueAt func(int)
 
 func (e *ByteArrayEncoding) DecodeByteArray(dst, src []byte) ([]byte, error) {
 	dst = dst[:0]
-	err := e.decode(src, func(value []byte) error {
-		dst = plain.AppendByteArray(dst, value)
-		return nil
+	err := e.decode(src, func(prefix, suffix []byte) ([]byte, error) {
+		n := len(prefix) + len(suffix)
+		b := [4]byte{}
+		plain.PutByteArrayLength(b[:], n)
+		dst = append(dst, b[:]...)
+		i := len(dst)
+		dst = append(dst, prefix...)
+		dst = append(dst, suffix...)
+		return dst[i:len(dst):len(dst)], nil
 	})
 	if err != nil {
 		err = encoding.Error(e, err)
@@ -116,12 +122,15 @@ func (e *ByteArrayEncoding) DecodeFixedLenByteArray(dst, src []byte, size int) (
 		return dst[:0], encoding.Error(e, encoding.ErrInvalidArgument)
 	}
 	dst = dst[:0]
-	err := e.decode(src, func(value []byte) error {
-		if len(value) != size {
-			return fmt.Errorf("cannot decode value of size %d into fixed-length byte array of size %d", len(value), size)
+	err := e.decode(src, func(prefix, suffix []byte) ([]byte, error) {
+		n := len(prefix) + len(suffix)
+		if n != size {
+			return nil, fmt.Errorf("cannot decode value of size %d into fixed-length byte array of size %d", n, size)
 		}
-		dst = append(dst, value...)
-		return nil
+		i := len(dst)
+		dst = append(dst, prefix...)
+		dst = append(dst, suffix...)
+		return dst[i:len(dst):len(dst)], nil
 	})
 	if err != nil {
 		err = encoding.Error(e, err)
@@ -129,7 +138,7 @@ func (e *ByteArrayEncoding) DecodeFixedLenByteArray(dst, src []byte, size int) (
 	return dst, err
 }
 
-func (e *ByteArrayEncoding) decode(src []byte, observe func([]byte) error) error {
+func (e *ByteArrayEncoding) decode(src []byte, observe func(prefix, suffix []byte) ([]byte, error)) error {
 	prefix := getInt32Buffer()
 	defer putInt32Buffer(prefix)
 
@@ -149,9 +158,7 @@ func (e *ByteArrayEncoding) decode(src []byte, observe func([]byte) error) error
 		return fmt.Errorf("number of prefix and lengths mismatch: %d != %d", len(prefix.values), len(length.values))
 	}
 
-	value := getBytesBuffer()
-	defer putBytesBuffer(value)
-
+	var lastValue []byte
 	for i, n := range length.values {
 		if int(n) < 0 {
 			return fmt.Errorf("invalid negative value length: %d", n)
@@ -164,15 +171,15 @@ func (e *ByteArrayEncoding) decode(src []byte, observe func([]byte) error) error
 		if int(p) < 0 {
 			return fmt.Errorf("invalid negative prefix length: %d", p)
 		}
-		if int(p) > value.Len() {
-			return fmt.Errorf("prefix length %d is larger than the last value of size %d", p, value.Len())
+		if int(p) > len(lastValue) {
+			return fmt.Errorf("prefix length %d is larger than the last value of size %d", p, len(lastValue))
 		}
 
-		value.Truncate(int(p))
-		value.Write(src[:n])
-		src = src[n:]
+		prefix := lastValue[:p:p]
+		suffix := src[:n:n]
+		src = src[n:len(src):len(src)]
 
-		if err := observe(value.Bytes()); err != nil {
+		if lastValue, err = observe(prefix, suffix); err != nil {
 			return err
 		}
 	}
diff --git a/encoding/delta/delta.go b/encoding/delta/delta.go
index 9191411..e2168a1 100644
--- a/encoding/delta/delta.go
+++ b/encoding/delta/delta.go
@@ -1,7 +1,6 @@
 package delta
 
 import (
-	"bytes"
 	"sync"
 )
 
@@ -20,7 +19,6 @@ func (buf *int32Buffer) decode(src []byte) ([]byte, error) {
 
 var (
 	int32BufferPool sync.Pool // *int32Buffer
-	bytesBufferPool sync.Pool // *bytes.Buffer
 )
 
 func getInt32Buffer() *int32Buffer {
@@ -38,17 +36,3 @@ func getInt32Buffer() *int32Buffer {
 func putInt32Buffer(b *int32Buffer) {
 	int32BufferPool.Put(b)
 }
-
-func getBytesBuffer() *bytes.Buffer {
-	b, _ := bytesBufferPool.Get().(*bytes.Buffer)
-	if b != nil {
-		b.Reset()
-	} else {
-		b = new(bytes.Buffer)
-	}
-	return b
-}
-
-func putBytesBuffer(b *bytes.Buffer) {
-	bytesBufferPool.Put(b)
-}
diff --git a/page.go b/page.go
index 913d8b0..3b4d4f4 100644
--- a/page.go
+++ b/page.go
@@ -444,27 +444,35 @@ func (page *repeatedPage) Slice(i, j int64) BufferedPage {
 		panic(errPageBoundsOutOfRange(i, j, numRows))
 	}
 
-	rowIndex0 := int64(0)
-	rowIndex1 := int64(len(page.repetitionLevels))
-	rowIndex2 := int64(len(page.repetitionLevels))
+	rowIndex0 := 0
+	rowIndex1 := len(page.repetitionLevels)
+	rowIndex2 := len(page.repetitionLevels)
 
 	for k, def := range page.repetitionLevels {
-		if def != page.maxRepetitionLevel {
-			if rowIndex0 == i {
-				rowIndex1 = int64(k)
+		if def == 0 {
+			if rowIndex0 == int(i) {
+				rowIndex1 = k
+				break
 			}
-			if rowIndex0 == j {
-				rowIndex2 = int64(k)
+			rowIndex0++
+		}
+	}
+
+	for k, def := range page.repetitionLevels[rowIndex1:] {
+		if def == 0 {
+			if rowIndex0 == int(j) {
+				rowIndex2 = rowIndex1 + k
+				break
 			}
 			rowIndex0++
 		}
 	}
 
-	numNulls1 := int64(countLevelsNotEqual(page.definitionLevels[:rowIndex1], page.maxDefinitionLevel))
-	numNulls2 := int64(countLevelsNotEqual(page.definitionLevels[rowIndex1:rowIndex2], page.maxDefinitionLevel))
+	numNulls1 := countLevelsNotEqual(page.definitionLevels[:rowIndex1], page.maxDefinitionLevel)
+	numNulls2 := countLevelsNotEqual(page.definitionLevels[rowIndex1:rowIndex2], page.maxDefinitionLevel)
 
-	i = rowIndex1 - numNulls1
-	j = rowIndex2 - (numNulls1 + numNulls2)
+	i = int64(rowIndex1 - numNulls1)
+	j = int64(rowIndex2 - (numNulls1 + numNulls2))
 
 	return newRepeatedPage(
 		page.base.Slice(i, j),
diff --git a/print.go b/print.go
index ea2b930..92d47d5 100644
--- a/print.go
+++ b/print.go
@@ -223,7 +223,6 @@ func PrintRowGroup(w io.Writer, rowGroup RowGroup) error {
 
 	rowbuf := make([]Row, defaultRowBufferSize)
 	cells := make([]string, 0, len(columns))
-	parts := make([]string, 0)
 	rows := rowGroup.Rows()
 	defer rows.Close()
 
@@ -233,25 +232,19 @@ func PrintRowGroup(w io.Writer, rowGroup RowGroup) error {
 		for _, row := range rowbuf[:n] {
 			cells = cells[:0]
 
-			for i := 0; i < len(row); {
-				j := i + 1
+			for _, value := range row {
+				columnIndex := value.Column()
 
-				for j < len(row) && row[j].Column() == row[i].Column() {
-					j++
+				for len(cells) <= columnIndex {
+					cells = append(cells, "")
 				}
 
-				if (j - i) == 1 {
-					cells = append(cells, row[i].String())
+				if cells[columnIndex] == "" {
+					cells[columnIndex] = value.String()
 				} else {
-					parts = parts[:0]
-					for k := i; k < j; k++ {
-						parts = append(parts, row[k].String())
-					}
-					alignment[len(cells)] = tablewriter.ALIGN_LEFT
-					cells = append(cells, strings.Join(parts, ","))
+					cells[columnIndex] += "," + value.String()
+					alignment[columnIndex] = tablewriter.ALIGN_LEFT
 				}
-
-				i = j
 			}
 
 			tw.Append(cells)
diff --git a/row.go b/row.go
index bf051ea..adf2a90 100644
--- a/row.go
+++ b/row.go
@@ -568,7 +568,7 @@ func reconstructFuncOfRepeated(columnIndex int16, node Node) (int16, reconstruct
 
 func reconstructRepeated(columnIndex, rowLength int16, levels levels, row Row, do func(levels, Row) (Row, error)) (Row, error) {
 	if !row.startsWith(columnIndex) {
-		return row, fmt.Errorf("row is missing repeated column %d", columnIndex)
+		return row, fmt.Errorf("row is missing repeated column %d: %+v", columnIndex, row)
 	}
 	if len(row) < int(rowLength) {
 		return row, fmt.Errorf("expected repeated column %d to have at least %d values but got %d", columnIndex, rowLength, len(row))


Test patch:
diff --git a/buffer_test.go b/buffer_test.go
index faaffa0..9b5154e 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -8,6 +8,7 @@ import (
 	"math/rand"
 	"reflect"
 	"sort"
+	"strconv"
 	"testing"
 	"testing/quick"
 	"time"
@@ -508,6 +509,50 @@ func TestBufferRoundtripNestedRepeatedPointer(t *testing.T) {
 	}
 }
 
+func TestBufferSeekToRow(t *testing.T) {
+	type B struct {
+		I int
+		C []string
+	}
+	type A struct {
+		B []B
+	}
+
+	buffer := parquet.NewBuffer()
+	var objs []A
+	for i := 0; i < 2; i++ {
+		o := A{
+			B: []B{
+				{I: i, C: []string{"foo", strconv.Itoa(i)}},
+				{I: i + 1, C: []string{"bar", strconv.Itoa(i + 1)}},
+			},
+		}
+		buffer.Write(&o)
+		objs = append(objs, o)
+	}
+
+	buf := new(bytes.Buffer)
+	w := parquet.NewWriter(buf)
+	w.WriteRowGroup(buffer)
+	w.Flush()
+	w.Close()
+
+	file := bytes.NewReader(buf.Bytes())
+	r := parquet.NewReader(file)
+
+	i := 1
+	o := new(A)
+	if err := r.SeekToRow(int64(i)); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Read(o); err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(*o, objs[i]) {
+		t.Errorf("points mismatch at row index %d: want=%v got=%v", i, objs[i], o)
+	}
+}
+
 type benchmarkBufferRowType struct {
 	ID    [16]byte `parquet:"id,uuid"`
 	Value float64  `parquet:"value"`
diff --git a/encoding/bytestreamsplit/bytestreamsplit.go b/encoding/bytestreamsplit/bytestreamsplit.go
index 6dc5b16..bc4e2a5 100644
--- a/encoding/bytestreamsplit/bytestreamsplit.go
+++ b/encoding/bytestreamsplit/bytestreamsplit.go
@@ -3,7 +3,6 @@ package bytestreamsplit
 import (
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
-	"github.com/segmentio/parquet-go/internal/bits"
 )
 
 // This encoder implements a version of the Byte Stream Split encoding as described
@@ -24,21 +23,8 @@ func (e *Encoding) EncodeFloat(dst, src []byte) ([]byte, error) {
 	if (len(src) % 4) != 0 {
 		return dst[:0], encoding.ErrEncodeInvalidInputSize(e, "FLOAT", len(src))
 	}
-
 	dst = resize(dst, len(src))
-	n := len(src) / 4
-	b0 := dst[0*n : 1*n]
-	b1 := dst[1*n : 2*n]
-	b2 := dst[2*n : 3*n]
-	b3 := dst[3*n : 4*n]
-
-	for i, v := range bits.BytesToUint32(src) {
-		b0[i] = byte(v >> 0)
-		b1[i] = byte(v >> 8)
-		b2[i] = byte(v >> 16)
-		b3[i] = byte(v >> 24)
-	}
-
+	encodeFloat(dst, src)
 	return dst, nil
 }
 
@@ -46,29 +32,8 @@ func (e *Encoding) EncodeDouble(dst, src []byte) ([]byte, error) {
 	if (len(src) % 8) != 0 {
 		return dst[:0], encoding.ErrEncodeInvalidInputSize(e, "DOUBLE", len(src))
 	}
-
 	dst = resize(dst, len(src))
-	n := len(src) / 8
-	b0 := dst[0*n : 1*n]
-	b1 := dst[1*n : 2*n]
-	b2 := dst[2*n : 3*n]
-	b3 := dst[3*n : 4*n]
-	b4 := dst[4*n : 5*n]
-	b5 := dst[5*n : 6*n]
-	b6 := dst[6*n : 7*n]
-	b7 := dst[7*n : 8*n]
-
-	for i, v := range bits.BytesToUint64(src) {
-		b0[i] = byte(v >> 0)
-		b1[i] = byte(v >> 8)
-		b2[i] = byte(v >> 16)
-		b3[i] = byte(v >> 24)
-		b4[i] = byte(v >> 32)
-		b5[i] = byte(v >> 40)
-		b6[i] = byte(v >> 48)
-		b7[i] = byte(v >> 56)
-	}
-
+	encodeDouble(dst, src)
 	return dst, nil
 }
 
@@ -76,22 +41,8 @@ func (e *Encoding) DecodeFloat(dst, src []byte) ([]byte, error) {
 	if (len(src) % 4) != 0 {
 		return dst[:0], encoding.ErrDecodeInvalidInputSize(e, "FLOAT", len(src))
 	}
-
 	dst = resize(dst, len(src))
-	n := len(src) / 4
-	b0 := src[0*n : 1*n]
-	b1 := src[1*n : 2*n]
-	b2 := src[2*n : 3*n]
-	b3 := src[3*n : 4*n]
-
-	dst32 := bits.BytesToUint32(dst)
-	for i := range dst32 {
-		dst32[i] = uint32(b0[i]) |
-			uint32(b1[i])<<8 |
-			uint32(b2[i])<<16 |
-			uint32(b3[i])<<24
-	}
-
+	decodeFloat(dst, src)
 	return dst, nil
 }
 
@@ -99,30 +50,8 @@ func (e *Encoding) DecodeDouble(dst, src []byte) ([]byte, error) {
 	if (len(src) % 8) != 0 {
 		return dst[:0], encoding.ErrDecodeInvalidInputSize(e, "DOUBLE", len(src))
 	}
-
 	dst = resize(dst, len(src))
-	n := len(src) / 8
-	b0 := src[0*n : 1*n]
-	b1 := src[1*n : 2*n]
-	b2 := src[2*n : 3*n]
-	b3 := src[3*n : 4*n]
-	b4 := src[4*n : 5*n]
-	b5 := src[5*n : 6*n]
-	b6 := src[6*n : 7*n]
-	b7 := src[7*n : 8*n]
-
-	dst64 := bits.BytesToUint64(dst)
-	for i := range dst64 {
-		dst64[i] = uint64(b0[i]) |
-			uint64(b1[i])<<8 |
-			uint64(b2[i])<<16 |
-			uint64(b3[i])<<24 |
-			uint64(b4[i])<<32 |
-			uint64(b5[i])<<40 |
-			uint64(b6[i])<<48 |
-			uint64(b7[i])<<56
-	}
-
+	decodeDouble(dst, src)
 	return dst, nil
 }
 
diff --git a/encoding/bytestreamsplit/bytestreamsplit_amd64.go b/encoding/bytestreamsplit/bytestreamsplit_amd64.go
new file mode 100644
index 0000000..1798c89
--- /dev/null
+++ b/encoding/bytestreamsplit/bytestreamsplit_amd64.go
@@ -0,0 +1,35 @@
+//go:build !purego
+
+package bytestreamsplit
+
+import (
+	"golang.org/x/sys/cpu"
+)
+
+var encodeFloatHasAVX512 = cpu.X86.HasAVX512 &&
+	cpu.X86.HasAVX512F &&
+	cpu.X86.HasAVX512VL
+
+var encodeDoubleHasAVX512 = cpu.X86.HasAVX512 &&
+	cpu.X86.HasAVX512F &&
+	cpu.X86.HasAVX512VL &&
+	cpu.X86.HasAVX512VBMI // VPERMB
+
+var decodeFloatHasAVX2 = cpu.X86.HasAVX2
+
+var decodeDoubleHasAVX512 = cpu.X86.HasAVX512 &&
+	cpu.X86.HasAVX512F &&
+	cpu.X86.HasAVX512VL &&
+	cpu.X86.HasAVX512VBMI // VPERMB
+
+//go:noescape
+func encodeFloat(dst, src []byte)
+
+//go:noescape
+func encodeDouble(dst, src []byte)
+
+//go:noescape
+func decodeFloat(dst, src []byte)
+
+//go:noescape
+func decodeDouble(dst, src []byte)
diff --git a/encoding/bytestreamsplit/bytestreamsplit_amd64.s b/encoding/bytestreamsplit/bytestreamsplit_amd64.s
new file mode 100644
index 0000000..b0c7622
--- /dev/null
+++ b/encoding/bytestreamsplit/bytestreamsplit_amd64.s
@@ -0,0 +1,426 @@
+ //go:build !purego
+
+#include "textflag.h"
+
+// This file contains optimizations of the BYTE_STREAM_SPLIT encoding using AVX2
+// and AVX512 (when available).
+//
+// The AVX2/512 instruction set comes with instructions to load memory from, or
+// store memory at sparse locations called VPGATHER and VPSCATTER. VPGATHER was
+// available in the AVX2 instruction set, VPSCATTER was introduced in AVX512
+// (when the AVX512_VBMI extension is supported). Gathering bytes are sparse
+// memory locations is useful during the decoding process since we are
+// recomposing 32 or 64 bit floating point values from 4 or 8 bytes dispatched
+// in the input byte array.
+//
+// To either deconstruct or reconstruct floating point values, we need to
+// reorder the bytes of each value. If we have 4 32 bit floats, we can permute
+// their bytes so that the first one contains all the first bytes, the second
+// contains all the second bytes, etc... The VPSHUFB instruction is used to
+// perform the byte permutation, or the VPERMB instruction for 64 bit floats.
+//
+// We use different instructions because the VPSHUFB instruction works on two
+// lanes of 16 bytes when used on YMM registers. 4 32 bit floats take 16 bytes,
+// so a a YMM register can hold two lanes of 4 32 bit floats and the VPSHUFB
+// can permute the two sets of values in a single invocation. For 64 bit floats
+// we need to permute 8 values, which take 64 bytes and therefore need to be
+// held in a ZMM register and apply permutations across the entire register,
+// which is only possible using VPERMB.
+//
+// Technically we could use ZMM registers when working on 32 bit floats to work
+// on 16 values per iteration. However, measurements indicated that the latency
+// of VPGATHERDD/VPSCATTERDD on ZMM registers did not provide any improvements
+// to the throughput of the algorithms, but working on more values increased the
+// code complexity. Using YMM registers offered the best balance between
+// performance and maintainability.
+//
+// At a high level the vectorized algorithms are the following:
+//
+// encoding
+// --------
+//   * Load a vector of data from the input buffer
+//   * Permute bytes, grouping bytes by index
+//   * Scatter bytes of the register to the output buffer
+//
+// decoding
+// --------
+//   * Gather sparse bytes from the input buffer
+//   * Permute bytes, reconstructing the original values
+//   * Store the vector in the output buffer
+//
+// When AVX instructions are not available, the functions fallback to scalar
+// implementations of the algorithms. These yield much lower throughput, but
+// performed 20-30% better than the code generated by the Go compiler.
+
+// func encodeFloat(dst, src []byte)
+TEXT ·encodeFloat(SB), NOSPLIT, $0-48
+    MOVQ src_base+24(FP), AX
+    MOVQ src_len+32(FP), BX
+    MOVQ dst_base+0(FP), DX
+
+    MOVQ AX, CX
+    ADDQ BX, CX // end
+    SHRQ $2, BX // len
+
+    CMPQ BX, $0
+    JE done
+
+    CMPB ·encodeFloatHasAVX512(SB), $0
+    JE loop1x4
+
+    CMPQ BX, $8
+    JB loop1x4
+
+    MOVQ CX, DI
+    SUBQ AX, DI
+    SHRQ $5, DI
+    SHLQ $5, DI
+    ADDQ AX, DI
+
+    VMOVDQU32 shuffle8x4<>(SB), Y0
+    VPBROADCASTD BX, Y2
+    VPMULLD scale8x4<>(SB), Y2, Y2
+    VPADDD offset8x4<>(SB), Y2, Y2
+loop8x4:
+    KXORQ K1, K1, K1
+    KNOTQ K1, K1
+
+    VMOVDQU32 (AX), Y1
+    VPSHUFB Y0, Y1, Y1
+    VPSCATTERDD Y1, K1, (DX)(Y2*1)
+
+    ADDQ $32, AX
+    ADDQ $8, DX
+    CMPQ AX, DI
+    JNE loop8x4
+    VZEROUPPER
+
+    CMPQ AX, CX
+    JE done
+loop1x4:
+    MOVL (AX), SI
+    MOVQ DX, DI
+
+    MOVB SI, (DI)
+    SHRL $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRL $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRL $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+
+    ADDQ $4, AX
+    INCQ DX
+    CMPQ AX, CX
+    JB loop1x4
+done:
+    RET
+
+// func encodeDouble(dst, src []byte)
+TEXT ·encodeDouble(SB), NOSPLIT, $0-48
+    MOVQ src_base+24(FP), AX
+    MOVQ src_len+32(FP), BX
+    MOVQ dst_base+0(FP), DX
+
+    MOVQ AX, CX
+    ADDQ BX, CX
+    SHRQ $3, BX
+
+    CMPQ BX, $0
+    JE done
+
+    CMPB ·encodeDoubleHasAVX512(SB), $0
+    JE loop1x8
+
+    CMPQ BX, $8
+    JB loop1x8
+
+    MOVQ CX, DI
+    SUBQ AX, DI
+    SHRQ $6, DI
+    SHLQ $6, DI
+    ADDQ AX, DI
+
+    VMOVDQU64 shuffle8x8<>(SB), Z0
+    VPBROADCASTQ BX, Z2
+    VPMULLQ scale8x8<>(SB), Z2, Z2
+loop8x8:
+    KXORQ K1, K1, K1
+    KNOTQ K1, K1
+
+    VMOVDQU64 (AX), Z1
+    VPERMB Z1, Z0, Z1
+    VPSCATTERQQ Z1, K1, (DX)(Z2*1)
+
+    ADDQ $64, AX
+    ADDQ $8, DX
+    CMPQ AX, DI
+    JNE loop8x8
+    VZEROUPPER
+
+    CMPQ AX, CX
+    JE done
+loop1x8:
+    MOVQ (AX), SI
+    MOVQ DX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+    SHRQ $8, SI
+    ADDQ BX, DI
+
+    MOVB SI, (DI)
+
+    ADDQ $8, AX
+    INCQ DX
+    CMPQ AX, CX
+    JB loop1x8
+done:
+    RET
+
+// func decodeFloat(dst, src []byte)
+TEXT ·decodeFloat(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ dst_len+8(FP), BX
+    MOVQ src_base+24(FP), DX
+
+    MOVQ AX, CX
+    ADDQ BX, CX // end
+    SHRQ $2, BX // len
+
+    CMPQ BX, $0
+    JE done
+
+    CMPB ·decodeFloatHasAVX2(SB), $0
+    JE loop1x4
+
+    CMPQ BX, $8
+    JB loop1x4
+
+    MOVQ CX, DI
+    SUBQ AX, DI
+    SHRQ $5, DI
+    SHLQ $5, DI
+    ADDQ AX, DI
+
+    MOVQ $0xFFFFFFFF, SI
+    MOVQ BX, X5
+    MOVQ SI, X6
+    VMOVDQU shuffle8x4<>(SB), Y0
+    VPBROADCASTD X5, Y2
+    VPBROADCASTD X6, Y3
+    VPMULLD scale8x4<>(SB), Y2, Y2
+    VPADDD offset8x4<>(SB), Y2, Y2
+    VMOVDQU Y3, Y4
+loop8x4:
+    VPGATHERDD Y4, (DX)(Y2*1), Y1
+    VPSHUFB Y0, Y1, Y1
+    VMOVDQU Y1, (AX)
+    VMOVDQU Y3, Y4
+
+    ADDQ $32, AX
+    ADDQ $8, DX
+    CMPQ AX, DI
+    JNE loop8x4
+    VZEROUPPER
+
+    CMPQ AX, CX
+    JE done
+loop1x4:
+    MOVQ DX, DI
+    MOVBLZX (DI), R8
+    ADDQ BX, DI
+    MOVBLZX (DI), R9
+    ADDQ BX, DI
+    MOVBLZX (DI), R10
+    ADDQ BX, DI
+    MOVBLZX (DI), R11
+
+    SHLL $8, R9
+    SHLL $16, R10
+    SHLL $24, R11
+
+    ORL R9, R8
+    ORL R10, R8
+    ORL R11, R8
+
+    MOVL R8, (AX)
+
+    ADDQ $4, AX
+    INCQ DX
+    CMPQ AX, CX
+    JB loop1x4
+done:
+    RET
+
+// func decodeDouble(dst, src []byte)
+TEXT ·decodeDouble(SB), NOSPLIT, $0-48
+    MOVQ dst_base+0(FP), AX
+    MOVQ dst_len+8(FP), BX
+    MOVQ src_base+24(FP), DX
+
+    MOVQ AX, CX
+    ADDQ BX, CX
+    SHRQ $3, BX
+
+    CMPQ BX, $0
+    JE done
+
+    CMPB ·decodeDoubleHasAVX512(SB), $0
+    JE loop1x8
+
+    CMPQ BX, $8
+    JB loop1x8
+
+    MOVQ CX, DI
+    SUBQ AX, DI
+    SHRQ $6, DI
+    SHLQ $6, DI
+    ADDQ AX, DI
+
+    VMOVDQU64 shuffle8x8<>(SB), Z0
+    VPBROADCASTQ BX, Z2
+    VPMULLQ scale8x8<>(SB), Z2, Z2
+loop8x8:
+    KXORQ K1, K1, K1
+    KNOTQ K1, K1
+
+    VPGATHERQQ (DX)(Z2*1), K1, Z1
+    VPERMB Z1, Z0, Z1
+    VMOVDQU64 Z1, (AX)
+
+    ADDQ $64, AX
+    ADDQ $8, DX
+    CMPQ AX, DI
+    JNE loop8x8
+    VZEROUPPER
+
+    CMPQ AX, CX
+    JE done
+loop1x8:
+    MOVQ DX, DI
+    XORQ R12, R12
+
+    MOVBQZX (DI), R8
+    ADDQ BX, DI
+    MOVBQZX (DI), R9
+    ADDQ BX, DI
+    MOVBQZX (DI), R10
+    ADDQ BX, DI
+    MOVBQZX (DI), R11
+    ADDQ BX, DI
+
+    SHLQ $8, R9
+    SHLQ $16, R10
+    SHLQ $24, R11
+
+    ORQ R8, R12
+    ORQ R9, R12
+    ORQ R10, R12
+    ORQ R11, R12
+
+    MOVBQZX (DI), R8
+    ADDQ BX, DI
+    MOVBQZX (DI), R9
+    ADDQ BX, DI
+    MOVBQZX (DI), R10
+    ADDQ BX, DI
+    MOVBQZX (DI), R11
+
+    SHLQ $32, R8
+    SHLQ $40, R9
+    SHLQ $48, R10
+    SHLQ $56, R11
+
+    ORQ R8, R12
+    ORQ R9, R12
+    ORQ R10, R12
+    ORQ R11, R12
+
+    MOVQ R12, (AX)
+
+    ADDQ $8, AX
+    INCQ DX
+    CMPQ AX, CX
+    JB loop1x8
+done:
+    RET
+
+GLOBL scale8x4<>(SB), RODATA|NOPTR, $32
+DATA scale8x4<>+0(SB)/4,  $0
+DATA scale8x4<>+4(SB)/4,  $1
+DATA scale8x4<>+8(SB)/4,  $2
+DATA scale8x4<>+12(SB)/4, $3
+DATA scale8x4<>+16(SB)/4, $0
+DATA scale8x4<>+20(SB)/4, $1
+DATA scale8x4<>+24(SB)/4, $2
+DATA scale8x4<>+28(SB)/4, $3
+
+GLOBL offset8x4<>(SB), RODATA|NOPTR, $32
+DATA offset8x4<>+0(SB)/4,  $0
+DATA offset8x4<>+4(SB)/4,  $0
+DATA offset8x4<>+8(SB)/4,  $0
+DATA offset8x4<>+12(SB)/4, $0
+DATA offset8x4<>+16(SB)/4, $4
+DATA offset8x4<>+20(SB)/4, $4
+DATA offset8x4<>+24(SB)/4, $4
+DATA offset8x4<>+28(SB)/4, $4
+
+GLOBL shuffle8x4<>(SB), RODATA|NOPTR, $32
+DATA shuffle8x4<>+0(SB)/4,  $0x0C080400
+DATA shuffle8x4<>+4(SB)/4,  $0x0D090501
+DATA shuffle8x4<>+8(SB)/4,  $0x0E0A0602
+DATA shuffle8x4<>+12(SB)/4, $0x0F0B0703
+DATA shuffle8x4<>+16(SB)/4, $0x0C080400
+DATA shuffle8x4<>+20(SB)/4, $0x0D090501
+DATA shuffle8x4<>+24(SB)/4, $0x0E0A0602
+DATA shuffle8x4<>+28(SB)/4, $0x0F0B0703
+
+GLOBL scale8x8<>(SB), RODATA|NOPTR, $64
+DATA scale8x8<>+0(SB)/8,  $0
+DATA scale8x8<>+8(SB)/8,  $1
+DATA scale8x8<>+16(SB)/8, $2
+DATA scale8x8<>+24(SB)/8, $3
+DATA scale8x8<>+32(SB)/8, $4
+DATA scale8x8<>+40(SB)/8, $5
+DATA scale8x8<>+48(SB)/8, $6
+DATA scale8x8<>+56(SB)/8, $7
+
+GLOBL shuffle8x8<>(SB), RODATA|NOPTR, $64
+DATA shuffle8x8<>+0(SB)/8,  $0x3830282018100800
+DATA shuffle8x8<>+8(SB)/8,  $0x3931292119110901
+DATA shuffle8x8<>+16(SB)/8, $0x3A322A221A120A02
+DATA shuffle8x8<>+24(SB)/8, $0x3B332B231B130B03
+DATA shuffle8x8<>+32(SB)/8, $0x3C342C241C140C04
+DATA shuffle8x8<>+40(SB)/8, $0x3D352D251D150D05
+DATA shuffle8x8<>+48(SB)/8, $0x3E362E261E160E06
+DATA shuffle8x8<>+56(SB)/8, $0x3F372F271F170F07
diff --git a/encoding/bytestreamsplit/bytestreamsplit_purego.go b/encoding/bytestreamsplit/bytestreamsplit_purego.go
new file mode 100644
index 0000000..bad1bd1
--- /dev/null
+++ b/encoding/bytestreamsplit/bytestreamsplit_purego.go
@@ -0,0 +1,83 @@
+//go:build purego || !amd64
+
+package bytestreamsplit
+
+import "github.com/segmentio/parquet-go/internal/bits"
+
+func encodeFloat(dst, src []byte) {
+	n := len(src) / 4
+	b0 := dst[0*n : 1*n]
+	b1 := dst[1*n : 2*n]
+	b2 := dst[2*n : 3*n]
+	b3 := dst[3*n : 4*n]
+
+	for i, v := range bits.BytesToUint32(src) {
+		b0[i] = byte(v >> 0)
+		b1[i] = byte(v >> 8)
+		b2[i] = byte(v >> 16)
+		b3[i] = byte(v >> 24)
+	}
+}
+
+func encodeDouble(dst, src []byte) {
+	n := len(src) / 8
+	b0 := dst[0*n : 1*n]
+	b1 := dst[1*n : 2*n]
+	b2 := dst[2*n : 3*n]
+	b3 := dst[3*n : 4*n]
+	b4 := dst[4*n : 5*n]
+	b5 := dst[5*n : 6*n]
+	b6 := dst[6*n : 7*n]
+	b7 := dst[7*n : 8*n]
+
+	for i, v := range bits.BytesToUint64(src) {
+		b0[i] = byte(v >> 0)
+		b1[i] = byte(v >> 8)
+		b2[i] = byte(v >> 16)
+		b3[i] = byte(v >> 24)
+		b4[i] = byte(v >> 32)
+		b5[i] = byte(v >> 40)
+		b6[i] = byte(v >> 48)
+		b7[i] = byte(v >> 56)
+	}
+}
+
+func decodeFloat(dst, src []byte) {
+	n := len(src) / 4
+	b0 := src[0*n : 1*n]
+	b1 := src[1*n : 2*n]
+	b2 := src[2*n : 3*n]
+	b3 := src[3*n : 4*n]
+
+	dst32 := bits.BytesToUint32(dst)
+	for i := range dst32 {
+		dst32[i] = uint32(b0[i]) |
+			uint32(b1[i])<<8 |
+			uint32(b2[i])<<16 |
+			uint32(b3[i])<<24
+	}
+}
+
+func decodeDouble(dst, src []byte) {
+	n := len(src) / 8
+	b0 := src[0*n : 1*n]
+	b1 := src[1*n : 2*n]
+	b2 := src[2*n : 3*n]
+	b3 := src[3*n : 4*n]
+	b4 := src[4*n : 5*n]
+	b5 := src[5*n : 6*n]
+	b6 := src[6*n : 7*n]
+	b7 := src[7*n : 8*n]
+
+	dst64 := bits.BytesToUint64(dst)
+	for i := range dst64 {
+		dst64[i] = uint64(b0[i]) |
+			uint64(b1[i])<<8 |
+			uint64(b2[i])<<16 |
+			uint64(b3[i])<<24 |
+			uint64(b4[i])<<32 |
+			uint64(b5[i])<<40 |
+			uint64(b6[i])<<48 |
+			uint64(b7[i])<<56
+	}
+}
diff --git a/encoding/bytestreamsplit/bytestreamsplit_test.go b/encoding/bytestreamsplit/bytestreamsplit_test.go
index a40c0cb..389b823 100644
--- a/encoding/bytestreamsplit/bytestreamsplit_test.go
+++ b/encoding/bytestreamsplit/bytestreamsplit_test.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/segmentio/parquet-go/encoding/bytestreamsplit"
 	"github.com/segmentio/parquet-go/encoding/fuzz"
+	"github.com/segmentio/parquet-go/encoding/test"
 )
 
 func FuzzEncodeFloat(f *testing.F) {
@@ -17,3 +18,11 @@ func FuzzEncodeFloat(f *testing.F) {
 func FuzzEncodeDouble(f *testing.F) {
 	fuzz.EncodeDouble(f, new(bytestreamsplit.Encoding))
 }
+
+func TestEncodeFloat(t *testing.T) {
+	test.EncodeFloat(t, new(bytestreamsplit.Encoding), 0, 100)
+}
+
+func TestEncodeDouble(t *testing.T) {
+	test.EncodeDouble(t, new(bytestreamsplit.Encoding), 0, 100)
+}
diff --git a/encoding/encoding_test.go b/encoding/encoding_test.go
index 948201a..fed2acf 100644
--- a/encoding/encoding_test.go
+++ b/encoding/encoding_test.go
@@ -7,6 +7,7 @@ import (
 	"math"
 	"math/rand"
 	"testing"
+	"time"
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
@@ -570,11 +571,11 @@ func benchmarkEncodeBoolean(b *testing.B, e encoding.Encoding) {
 	values := generateBooleanValues(benchmarkNumValues, newRand())
 	setBitWidth(e, 1)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeBoolean(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeBoolean(buffer, values)
+		})
 	})
-
-	b.SetBytes(1 * int64(len(values)))
 }
 
 func benchmarkEncodeLevels(b *testing.B, e encoding.Encoding) {
@@ -583,11 +584,11 @@ func benchmarkEncodeLevels(b *testing.B, e encoding.Encoding) {
 	values := generateLevelValues(benchmarkNumValues, newRand())
 	setBitWidth(e, bits.MaxLen8(bits.BytesToInt8(values)))
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeLevels(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeLevels(buffer, values)
+		})
 	})
-
-	b.SetBytes(1 * int64(len(values)))
 }
 
 func benchmarkEncodeInt32(b *testing.B, e encoding.Encoding) {
@@ -596,11 +597,11 @@ func benchmarkEncodeInt32(b *testing.B, e encoding.Encoding) {
 	values := generateInt32Values(benchmarkNumValues, newRand())
 	setBitWidth(e, bits.MaxLen32(bits.BytesToInt32(values)))
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeInt32(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeInt32(buffer, values)
+		})
 	})
-
-	b.SetBytes(4 * int64(len(values)))
 }
 
 func benchmarkEncodeInt64(b *testing.B, e encoding.Encoding) {
@@ -609,11 +610,11 @@ func benchmarkEncodeInt64(b *testing.B, e encoding.Encoding) {
 	values := generateInt64Values(benchmarkNumValues, newRand())
 	setBitWidth(e, bits.MaxLen64(bits.BytesToInt64(values)))
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeInt64(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeInt64(buffer, values)
+		})
 	})
-
-	b.SetBytes(8 * int64(len(values)))
 }
 
 func benchmarkEncodeFloat(b *testing.B, e encoding.Encoding) {
@@ -621,11 +622,11 @@ func benchmarkEncodeFloat(b *testing.B, e encoding.Encoding) {
 	buffer := make([]byte, 0)
 	values := generateFloatValues(benchmarkNumValues, newRand())
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeFloat(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeFloat(buffer, values)
+		})
 	})
-
-	b.SetBytes(4 * int64(len(values)))
 }
 
 func benchmarkEncodeDouble(b *testing.B, e encoding.Encoding) {
@@ -633,11 +634,11 @@ func benchmarkEncodeDouble(b *testing.B, e encoding.Encoding) {
 	buffer := make([]byte, 0)
 	values := generateDoubleValues(benchmarkNumValues, newRand())
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeDouble(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeDouble(buffer, values)
+		})
 	})
-
-	b.SetBytes(8 * int64(len(values)))
 }
 
 func benchmarkEncodeByteArray(b *testing.B, e encoding.Encoding) {
@@ -645,11 +646,11 @@ func benchmarkEncodeByteArray(b *testing.B, e encoding.Encoding) {
 	buffer := make([]byte, 0)
 	values := generateByteArrayValues(benchmarkNumValues, newRand())
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeByteArray(buffer, values)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeByteArray(buffer, values)
+		})
 	})
-
-	b.SetBytes(int64(len(values)))
 }
 
 func benchmarkEncodeFixedLenByteArray(b *testing.B, e encoding.Encoding) {
@@ -658,11 +659,11 @@ func benchmarkEncodeFixedLenByteArray(b *testing.B, e encoding.Encoding) {
 	buffer := make([]byte, 0)
 	values := generateFixedLenByteArrayValues(benchmarkNumValues, newRand(), size)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		buffer, _ = e.EncodeFixedLenByteArray(buffer, values, size)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			buffer, _ = e.EncodeFixedLenByteArray(buffer, values, size)
+		})
 	})
-
-	b.SetBytes(int64(len(values)))
 }
 
 func BenchmarkDecode(b *testing.B) {
@@ -720,11 +721,11 @@ func benchmarkDecodeBoolean(b *testing.B, e encoding.Encoding) {
 	setBitWidth(e, 1)
 	buffer, _ := e.EncodeBoolean(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeBoolean(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeBoolean(output, buffer)
+		})
 	})
-
-	b.SetBytes(1 * int64(len(values)))
 }
 
 func benchmarkDecodeLevels(b *testing.B, e encoding.Encoding) {
@@ -734,11 +735,11 @@ func benchmarkDecodeLevels(b *testing.B, e encoding.Encoding) {
 	setBitWidth(e, bits.MaxLen8(bits.BytesToInt8(values)))
 	buffer, _ := e.EncodeLevels(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeLevels(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeLevels(output, buffer)
+		})
 	})
-
-	b.SetBytes(1 * int64(len(values)))
 }
 
 func benchmarkDecodeInt32(b *testing.B, e encoding.Encoding) {
@@ -748,11 +749,11 @@ func benchmarkDecodeInt32(b *testing.B, e encoding.Encoding) {
 	setBitWidth(e, bits.MaxLen32(bits.BytesToInt32(values)))
 	buffer, _ := e.EncodeInt32(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeInt32(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeInt32(output, buffer)
+		})
 	})
-
-	b.SetBytes(4 * int64(len(values)))
 }
 
 func benchmarkDecodeInt64(b *testing.B, e encoding.Encoding) {
@@ -762,11 +763,11 @@ func benchmarkDecodeInt64(b *testing.B, e encoding.Encoding) {
 	setBitWidth(e, bits.MaxLen64(bits.BytesToInt64(values)))
 	buffer, _ := e.EncodeInt64(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeInt64(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeInt64(output, buffer)
+		})
 	})
-
-	b.SetBytes(8 * int64(len(values)))
 }
 
 func benchmarkDecodeFloat(b *testing.B, e encoding.Encoding) {
@@ -775,11 +776,11 @@ func benchmarkDecodeFloat(b *testing.B, e encoding.Encoding) {
 	output := make([]byte, 0)
 	buffer, _ := e.EncodeFloat(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeFloat(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeFloat(output, buffer)
+		})
 	})
-
-	b.SetBytes(4 * int64(len(values)))
 }
 
 func benchmarkDecodeDouble(b *testing.B, e encoding.Encoding) {
@@ -788,11 +789,11 @@ func benchmarkDecodeDouble(b *testing.B, e encoding.Encoding) {
 	output := make([]byte, 0)
 	buffer, _ := e.EncodeDouble(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeDouble(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeDouble(output, buffer)
+		})
 	})
-
-	b.SetBytes(8 * int64(len(values)))
 }
 
 func benchmarkDecodeByteArray(b *testing.B, e encoding.Encoding) {
@@ -801,11 +802,11 @@ func benchmarkDecodeByteArray(b *testing.B, e encoding.Encoding) {
 	output := make([]byte, 0)
 	buffer, _ := e.EncodeByteArray(nil, values)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeByteArray(output, buffer)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeByteArray(output, buffer)
+		})
 	})
-
-	b.SetBytes(int64(len(values)))
 }
 
 func benchmarkDecodeFixedLenByteArray(b *testing.B, e encoding.Encoding) {
@@ -815,11 +816,11 @@ func benchmarkDecodeFixedLenByteArray(b *testing.B, e encoding.Encoding) {
 	output := make([]byte, 0)
 	buffer, _ := e.EncodeFixedLenByteArray(nil, values, size)
 
-	benchmarkZeroAllocsPerRun(b, func() {
-		output, _ = e.DecodeFixedLenByteArray(output, buffer, size)
+	reportThroughput(b, benchmarkNumValues, len(values), func() {
+		benchmarkZeroAllocsPerRun(b, func() {
+			output, _ = e.DecodeFixedLenByteArray(output, buffer, size)
+		})
 	})
-
-	b.SetBytes(int64(len(values)))
 }
 
 func benchmarkZeroAllocsPerRun(b *testing.B, f func()) {
@@ -828,6 +829,14 @@ func benchmarkZeroAllocsPerRun(b *testing.B, f func()) {
 	}
 }
 
+func reportThroughput(b *testing.B, numValues, numBytes int, do func()) {
+	start := time.Now()
+	do()
+	seconds := time.Since(start).Seconds()
+	b.SetBytes(int64(numBytes))
+	b.ReportMetric(float64(b.N*numValues)/seconds, "value/s")
+}
+
 func generateBooleanValues(n int, r *rand.Rand) []byte {
 	values := make([]byte, n/8+1)
 	io.ReadFull(r, values)
diff --git a/encoding/test/test.go b/encoding/test/test.go
new file mode 100644
index 0000000..a022dce
--- /dev/null
+++ b/encoding/test/test.go
@@ -0,0 +1,72 @@
+//go:build go1.18
+
+package test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/internal/unsafecast"
+)
+
+func EncodeFloat(t *testing.T, enc encoding.Encoding, min, max int) {
+	t.Helper()
+	encode(t, enc, min, max,
+		encoding.Encoding.EncodeFloat,
+		encoding.Encoding.DecodeFloat,
+		func(i int) float32 { return float32(i) },
+	)
+}
+
+func EncodeDouble(t *testing.T, enc encoding.Encoding, min, max int) {
+	t.Helper()
+	encode(t, enc, min, max,
+		encoding.Encoding.EncodeDouble,
+		encoding.Encoding.DecodeDouble,
+		func(i int) float64 { return float64(i) },
+	)
+}
+
+type encodingMethod func(encoding.Encoding, []byte, []byte) ([]byte, error)
+
+func encode[T comparable](t *testing.T, enc encoding.Encoding, min, max int, encode, decode encodingMethod, valueOf func(int) T) {
+	t.Helper()
+
+	for k := min; k <= max; k++ {
+		t.Run(fmt.Sprintf("N=%d", k), func(t *testing.T) {
+			src := make([]T, k)
+			for i := range src {
+				src[i] = valueOf(i)
+			}
+
+			buf, err := encode(enc, nil, unsafecast.Slice[byte](src))
+			if err != nil {
+				t.Fatalf("encoding %d values: %v", k, err)
+			}
+
+			res, err := decode(enc, nil, buf)
+			if err != nil {
+				t.Fatalf("decoding %d values: %v", k, err)
+			}
+
+			if err := assertEqual(src, unsafecast.Slice[T](res)); err != nil {
+				t.Fatalf("testing %d values: %v", k, err)
+			}
+		})
+	}
+}
+
+func assertEqual[T comparable](want, got []T) error {
+	if len(want) != len(got) {
+		return fmt.Errorf("number of values mismatch: want=%d got=%d", len(want), len(got))
+	}
+
+	for i := range want {
+		if want[i] != got[i] {
+			return fmt.Errorf("values at index %d/%d mismatch: want=%+v got=%+v", i, len(want), want[i], got[i])
+		}
+	}
+
+	return nil
+}