Repository: segmentio/parquet-go
Pull request #188 (2022-05-23 07:59:41)

Optimize byte stream split encoding
I was interested to learn about how to used the VPGATHER* and VPSCATTER* instructions in x86, so I used the BYTE_STREAM_SPLIT encoding as a test case.

Overall, the instructions bring measurable throughput improvements compared to the scalar versions of the algorithms, but they have high latency and don't seem to benefit from CPU pipelining. I tried to unroll the loops but it did not bring any throughput improvements. 

The gains are nevertheless interesting, the assembly code using scalar instructions was 20-30% faster than what the Go compiler generated, but using the vector instruction yields unlocked a lot more efficiencies:

```
name                             old time/op    new time/op     delta
Encode/BYTE_STREAM_SPLIT/float     13.7µs ± 0%      4.3µs ± 0%   -68.41%  (p=0.000 n=10+9)
Encode/BYTE_STREAM_SPLIT/double    35.7µs ± 0%      5.5µs ± 1%   -84.63%  (p=0.000 n=9+10)
Decode/BYTE_STREAM_SPLIT/float     12.8µs ± 0%      2.1µs ± 0%   -83.65%  (p=0.000 n=8+8)
Decode/BYTE_STREAM_SPLIT/double    28.0µs ± 0%      3.4µs ± 1%   -87.95%  (p=0.000 n=8+10)

name                             old speed      new speed       delta
Encode/BYTE_STREAM_SPLIT/float   2.91GB/s ± 0%   9.22GB/s ± 0%  +216.54%  (p=0.000 n=10+9)
Encode/BYTE_STREAM_SPLIT/double  2.24GB/s ± 0%  14.57GB/s ± 1%  +550.60%  (p=0.000 n=9+10)
Decode/BYTE_STREAM_SPLIT/float   3.13GB/s ± 0%  19.15GB/s ± 0%  +511.38%  (p=0.000 n=8+10)
Decode/BYTE_STREAM_SPLIT/double  2.86GB/s ± 0%  23.73GB/s ± 1%  +730.11%  (p=0.000 n=8+10)

name                             old value/s    new value/s     delta
Encode/BYTE_STREAM_SPLIT/float       728M ± 0%      2305M ± 0%  +216.54%  (p=0.000 n=10+9)
Encode/BYTE_STREAM_SPLIT/double      280M ± 0%      1822M ± 1%  +550.60%  (p=0.000 n=9+10)
Decode/BYTE_STREAM_SPLIT/float       783M ± 0%      4789M ± 0%  +511.37%  (p=0.000 n=8+10)
Decode/BYTE_STREAM_SPLIT/double      357M ± 0%      2966M ± 1%  +730.09%  (p=0.000 n=8+10)
```


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/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
+}