Repository: segmentio/parquet-go
Pull request #66 (2022-02-15 20:41:44)

[O11Y-217] always use int64 to represent row indexes
This PR unifies external APIs to always use `int64` instead of `int` for row indexes.

pelletier: What's the motivation behind this change?
achille-roussel: Go makes no strong guarantees on the capacity of a `int` value, it's "at least 32 bits", and there may be more than 2 billion rows to in a parquet file (or set of merged files).

Patch:
diff --git a/column_buffer.go b/column_buffer.go
index f2049c2..443ba01 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -2,11 +2,14 @@ package parquet
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"sort"
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/encoding/plain"
+	"github.com/segmentio/parquet-go/internal/bits"
 )
 
 // ColumnBuffer is an interface representing columns of a row group.
@@ -293,10 +296,10 @@ func (col *optionalColumnBuffer) WriteValues(values []Value) (n int, err error)
 
 func (col *optionalColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	var idx = -1
 	var err error
@@ -311,11 +314,11 @@ func (col *optionalColumnBuffer) WriteRow(row Row) error {
 	return err
 }
 
-func (col *optionalColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *optionalColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	if index < 0 {
-		return row, errRowIndexOutOfBounds(index, len(col.definitionLevels))
+		return row, errRowIndexOutOfBounds(index, int64(len(col.definitionLevels)))
 	}
-	if index >= len(col.definitionLevels) {
+	if index >= int64(len(col.definitionLevels)) {
 		return row, io.EOF
 	}
 
@@ -325,7 +328,7 @@ func (col *optionalColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
 		var err error
 		var n = len(row)
 
-		if row, err = col.base.ReadRowAt(row, int(col.rows[index])); err != nil {
+		if row, err = col.base.ReadRowAt(row, int64(col.rows[index])); err != nil {
 			return row, err
 		}
 
@@ -447,11 +450,11 @@ func (col *repeatedColumnBuffer) Page() BufferedPage {
 
 		for _, row := range col.rows {
 			numNulls := countLevelsNotEqual(col.definitionLevels[row.offset:row.offset+row.length], col.maxDefinitionLevel)
-			numValues := int(row.length) - numNulls
+			numValues := int64(row.length) - int64(numNulls)
 
-			for i := 0; i < numValues; i++ {
+			for i := int64(0); i < numValues; i++ {
 				var err error
-				if buffer, err = col.base.ReadRowAt(buffer[:0], int(row.offset)+i); err != nil {
+				if buffer, err = col.base.ReadRowAt(buffer[:0], int64(row.offset)+i); err != nil {
 					return newErrorPage(col.Column(), "reordering rows of repeated column: %w", err)
 				}
 				if err = column.base.WriteRow(buffer); err != nil {
@@ -570,7 +573,7 @@ func (col *repeatedColumnBuffer) WriteValues(values []Value) (numValues int, err
 
 func (col *repeatedColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 
 	col.rows = append(col.rows, region{
@@ -596,11 +599,11 @@ func (col *repeatedColumnBuffer) WriteRow(row Row) error {
 	return nil
 }
 
-func (col *repeatedColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *repeatedColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	if index < 0 {
-		return row, errRowIndexOutOfBounds(index, len(col.rows))
+		return row, errRowIndexOutOfBounds(index, int64(len(col.rows)))
 	}
-	if index >= len(col.rows) {
+	if index >= int64(len(col.rows)) {
 		return row, io.EOF
 	}
 
@@ -609,7 +612,7 @@ func (col *repeatedColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
 	maxDefinitionLevel := col.maxDefinitionLevel
 	repetitionLevels := col.repetitionLevels[region.offset : region.offset+region.length]
 	definitionLevels := col.definitionLevels[region.offset : region.offset+region.length]
-	baseIndex := 0
+	baseIndex := int64(0)
 
 	for i := range definitionLevels {
 		if definitionLevels[i] != maxDefinitionLevel {
@@ -619,14 +622,14 @@ func (col *repeatedColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
 			})
 		} else {
 			var err error
-			var n = len(row)
+			var n = int64(len(row))
 
-			if row, err = col.base.ReadRowAt(row, int(region.offset)+baseIndex); err != nil {
+			if row, err = col.base.ReadRowAt(row, int64(region.offset)+baseIndex); err != nil {
 				return row[:reset], err
 			}
 
 			baseIndex += n
-			for n < len(row) {
+			for n < int64(len(row)) {
 				row[n].repetitionLevel = repetitionLevels[i]
 				row[n].definitionLevel = definitionLevels[i]
 				n++
@@ -726,6 +729,15 @@ func (col *booleanColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *booleanColumnBuffer) Write(b []byte) (int, error) {
+	return col.WriteBooleans(bits.BytesToBool(b))
+}
+
+func (col *booleanColumnBuffer) WriteBooleans(values []bool) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Boolean())
@@ -735,20 +747,20 @@ func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *booleanColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Boolean())
 	return nil
 }
 
-func (col *booleanColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *booleanColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueBoolean(col.values[index])
@@ -808,6 +820,19 @@ func (col *int32ColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *int32ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 4) != 0 {
+		return 0, fmt.Errorf("cannot write INT32 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToInt32(b)...)
+	return len(b), nil
+}
+
+func (col *int32ColumnBuffer) WriteInt32s(values []int32) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Int32())
@@ -817,20 +842,20 @@ func (col *int32ColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *int32ColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Int32())
 	return nil
 }
 
-func (col *int32ColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *int32ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueInt32(col.values[index])
@@ -890,6 +915,19 @@ func (col *int64ColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *int64ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 8) != 0 {
+		return 0, fmt.Errorf("cannot write INT64 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToInt64(b)...)
+	return len(b), nil
+}
+
+func (col *int64ColumnBuffer) WriteInt64s(values []int64) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Int64())
@@ -899,20 +937,20 @@ func (col *int64ColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *int64ColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Int64())
 	return nil
 }
 
-func (col *int64ColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *int64ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueInt64(col.values[index])
@@ -972,6 +1010,19 @@ func (col *int96ColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *int96ColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 12) != 0 {
+		return 0, fmt.Errorf("cannot write INT96 values from input of size %d", len(b))
+	}
+	col.values = append(col.values, deprecated.BytesToInt96(b)...)
+	return len(b), nil
+}
+
+func (col *int96ColumnBuffer) WriteInt96s(values []deprecated.Int96) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Int96())
@@ -981,20 +1032,20 @@ func (col *int96ColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *int96ColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Int96())
 	return nil
 }
 
-func (col *int96ColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *int96ColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueInt96(col.values[index])
@@ -1054,6 +1105,19 @@ func (col *floatColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *floatColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 4) != 0 {
+		return 0, fmt.Errorf("cannot write FLOAT values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToFloat32(b)...)
+	return len(b), nil
+}
+
+func (col *floatColumnBuffer) WriteFloats(values []float32) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Float())
@@ -1063,20 +1127,20 @@ func (col *floatColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *floatColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Float())
 	return nil
 }
 
-func (col *floatColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *floatColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueFloat(col.values[index])
@@ -1136,6 +1200,19 @@ func (col *doubleColumnBuffer) Swap(i, j int) {
 	col.values[i], col.values[j] = col.values[j], col.values[i]
 }
 
+func (col *doubleColumnBuffer) Write(b []byte) (int, error) {
+	if (len(b) % 8) != 0 {
+		return 0, fmt.Errorf("cannot write DOUBLE values from input of size %d", len(b))
+	}
+	col.values = append(col.values, bits.BytesToFloat64(b)...)
+	return len(b), nil
+}
+
+func (col *doubleColumnBuffer) WriteDoubles(values []float64) (int, error) {
+	col.values = append(col.values, values...)
+	return len(values), nil
+}
+
 func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values = append(col.values, v.Double())
@@ -1145,20 +1222,20 @@ func (col *doubleColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *doubleColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, row[0].Double())
 	return nil
 }
 
-func (col *doubleColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *doubleColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := makeValueDouble(col.values[index])
@@ -1220,6 +1297,26 @@ func (col *byteArrayColumnBuffer) Less(i, j int) bool { return col.values.Less(i
 
 func (col *byteArrayColumnBuffer) Swap(i, j int) { col.values.Swap(i, j) }
 
+func (col *byteArrayColumnBuffer) Write(b []byte) (int, error) {
+	_, n, err := col.writeByteArrays(b)
+	return n, err
+}
+
+func (col *byteArrayColumnBuffer) WriteByteArrays(values []byte) (int, error) {
+	n, _, err := col.writeByteArrays(values)
+	return n, err
+}
+
+func (col *byteArrayColumnBuffer) writeByteArrays(values []byte) (c, n int, err error) {
+	err = plain.RangeByteArrays(values, func(v []byte) error {
+		col.values.Push(v)
+		n += plain.ByteArrayLengthSize + len(v)
+		c++
+		return nil
+	})
+	return c, n, err
+}
+
 func (col *byteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.values.Push(v.ByteArray())
@@ -1229,23 +1326,23 @@ func (col *byteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *byteArrayColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values.Push(row[0].ByteArray())
 	return nil
 }
 
-func (col *byteArrayColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *byteArrayColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, col.values.Len())
-	case index >= col.values.Len():
+		return row, errRowIndexOutOfBounds(index, int64(col.values.Len()))
+	case index >= int64(col.values.Len()):
 		return row, io.EOF
 	default:
-		v := makeValueBytes(ByteArray, col.values.Index(index))
+		v := makeValueBytes(ByteArray, col.values.Index(int(index)))
 		v.columnIndex = col.columnIndex
 		return append(row, v), nil
 	}
@@ -1323,6 +1420,20 @@ func (col *fixedLenByteArrayColumnBuffer) index(i int) []byte {
 	return col.data[j:k:k]
 }
 
+func (col *fixedLenByteArrayColumnBuffer) Write(b []byte) (int, error) {
+	n, err := col.WriteFixedLenByteArrays(b)
+	return n * col.size, err
+}
+
+func (col *fixedLenByteArrayColumnBuffer) WriteFixedLenByteArrays(values []byte) (int, error) {
+	d, m := len(values)/col.size, len(values)%col.size
+	if m != 0 {
+		return 0, fmt.Errorf("cannot write FIXED_LEN_BYTE_ARRAY values of size %d from input of size %d", col.size, len(values))
+	}
+	col.data = append(col.data, values...)
+	return d, nil
+}
+
 func (col *fixedLenByteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
 		col.data = append(col.data, v.ByteArray()...)
@@ -1332,22 +1443,22 @@ func (col *fixedLenByteArrayColumnBuffer) WriteValues(values []Value) (int, erro
 
 func (col *fixedLenByteArrayColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.data = append(col.data, row[0].ByteArray()...)
 	return nil
 }
 
-func (col *fixedLenByteArrayColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
-	i := (index + 0) * col.size
-	j := (index + 1) * col.size
+func (col *fixedLenByteArrayColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
+	i := (index + 0) * int64(col.size)
+	j := (index + 1) * int64(col.size)
 	switch {
 	case i < 0:
-		return row, errRowIndexOutOfBounds(index, col.Len())
-	case j > len(col.data):
+		return row, errRowIndexOutOfBounds(index, int64(col.Len()))
+	case j > int64(len(col.data)):
 		return row, io.EOF
 	default:
 		v := makeValueBytes(FixedLenByteArray, col.data[i:j])
@@ -1406,4 +1517,22 @@ func (col uint64ColumnBuffer) Less(i, j int) bool {
 
 var (
 	_ sort.Interface = (ColumnBuffer)(nil)
+
+	_ io.Writer = (*booleanColumnBuffer)(nil)
+	_ io.Writer = (*int32ColumnBuffer)(nil)
+	_ io.Writer = (*int64ColumnBuffer)(nil)
+	_ io.Writer = (*int96ColumnBuffer)(nil)
+	_ io.Writer = (*floatColumnBuffer)(nil)
+	_ io.Writer = (*doubleColumnBuffer)(nil)
+	_ io.Writer = (*byteArrayColumnBuffer)(nil)
+	_ io.Writer = (*fixedLenByteArrayColumnBuffer)(nil)
+
+	_ BooleanWriter           = (*booleanColumnBuffer)(nil)
+	_ Int32Writer             = (*int32ColumnBuffer)(nil)
+	_ Int64Writer             = (*int64ColumnBuffer)(nil)
+	_ Int96Writer             = (*int96ColumnBuffer)(nil)
+	_ FloatWriter             = (*floatColumnBuffer)(nil)
+	_ DoubleWriter            = (*doubleColumnBuffer)(nil)
+	_ ByteArrayWriter         = (*byteArrayColumnBuffer)(nil)
+	_ FixedLenByteArrayWriter = (*fixedLenByteArrayColumnBuffer)(nil)
 )
diff --git a/deprecated/int96.go b/deprecated/int96.go
index dc26d38..e242f3c 100644
--- a/deprecated/int96.go
+++ b/deprecated/int96.go
@@ -71,6 +71,15 @@ func Int96ToBytes(data []Int96) []byte {
 	return unsafe.Slice(*(**byte)(unsafe.Pointer(&data)), 12*len(data))
 }
 
+// BytesToInt96 converts the byte slice passed as argument to a slice of Int96
+// sharing the same backing array.
+//
+// When the number of bytes in the input is not a multiple of 12, the function
+// truncates it in the returned slice.
+func BytesToInt96(data []byte) []Int96 {
+	return unsafe.Slice(*(**Int96)(unsafe.Pointer(&data)), len(data)/12)
+}
+
 func MaxLenInt96(data []Int96) int {
 	max := 0
 	for i := range data {
diff --git a/dictionary.go b/dictionary.go
index 1ed7d49..fa9dba3 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -839,20 +839,20 @@ func (col *indexedColumnBuffer) WriteValues(values []Value) (int, error) {
 
 func (col *indexedColumnBuffer) WriteRow(row Row) error {
 	if len(row) == 0 {
-		return errRowHasTooFewValues(len(row))
+		return errRowHasTooFewValues(int64(len(row)))
 	}
 	if len(row) > 1 {
-		return errRowHasTooManyValues(len(row))
+		return errRowHasTooManyValues(int64(len(row)))
 	}
 	col.values = append(col.values, int32(col.dict.Insert(row[0])))
 	return nil
 }
 
-func (col *indexedColumnBuffer) ReadRowAt(row Row, index int) (Row, error) {
+func (col *indexedColumnBuffer) ReadRowAt(row Row, index int64) (Row, error) {
 	switch {
 	case index < 0:
-		return row, errRowIndexOutOfBounds(index, len(col.values))
-	case index >= len(col.values):
+		return row, errRowIndexOutOfBounds(index, int64(len(col.values)))
+	case index >= int64(len(col.values)):
 		return row, io.EOF
 	default:
 		v := col.dict.Index(int(col.values[index]))
diff --git a/encoding/plain/plain.go b/encoding/plain/plain.go
index 8d7d5bf..65738b9 100644
--- a/encoding/plain/plain.go
+++ b/encoding/plain/plain.go
@@ -5,6 +5,7 @@ package plain
 
 import (
 	"encoding/binary"
+	"fmt"
 	"io"
 	"math"
 
@@ -13,6 +14,10 @@ import (
 	"github.com/segmentio/parquet-go/format"
 )
 
+const (
+	ByteArrayLengthSize = 4
+)
+
 type Encoding struct {
 }
 
@@ -48,6 +53,8 @@ func Float(v float32) []byte { return AppendFloat(nil, v) }
 
 func Double(v float64) []byte { return AppendDouble(nil, v) }
 
+func ByteArray(v []byte) []byte { return AppendByteArray(nil, v) }
+
 func AppendBoolean(b []byte, v bool) []byte {
 	if v {
 		b = append(b, 1)
@@ -88,3 +95,40 @@ func AppendDouble(b []byte, v float64) []byte {
 	binary.LittleEndian.PutUint64(x[:], math.Float64bits(v))
 	return append(b, x[:]...)
 }
+
+func AppendByteArray(b, v []byte) []byte {
+	i := len(b)
+	j := i + 4
+	b = append(b, 0, 0, 0, 0)
+	b = append(b, v...)
+	PutByteArrayLength(b[i:j:j], len(v))
+	return b
+}
+
+func PutByteArrayLength(b []byte, n int) {
+	binary.LittleEndian.PutUint32(b, uint32(n))
+}
+
+func RangeByteArrays(b []byte, do func([]byte) error) (err error) {
+	for len(b) > 0 {
+		var v []byte
+		if v, b, err = NextByteArray(b); err != nil {
+			return err
+		}
+		if err = do(v); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func NextByteArray(b []byte) (v, r []byte, err error) {
+	if len(b) < 4 {
+		return nil, b, fmt.Errorf("input of length %d is too short to contain a PLAIN encoded byte array: %w", len(b), io.ErrUnexpectedEOF)
+	}
+	n := 4 + int(binary.LittleEndian.Uint32(b))
+	if n > len(b) {
+		return nil, b, fmt.Errorf("input of length %d is too short to contain a PLAIN encoded byte array of length %d: %w", len(b)-4, n-4, io.ErrUnexpectedEOF)
+	}
+	return b[4:n], b[n:], nil
+}
diff --git a/internal/bits/bits_amd64.go b/internal/bits/bits_amd64.go
index 560e2de..22f0979 100644
--- a/internal/bits/bits_amd64.go
+++ b/internal/bits/bits_amd64.go
@@ -1,6 +1,8 @@
 package bits
 
-import "golang.org/x/sys/cpu"
+import (
+	"golang.org/x/sys/cpu"
+)
 
 // All functions require at least the F and VL instruction sets (Foundation and
 // Vector Length extensions). Note that Foundation is a requirement of AVX-512,
diff --git a/internal/bits/max_amd64.s b/internal/bits/max_amd64.s
index 6594c9c..3a152c4 100644
--- a/internal/bits/max_amd64.s
+++ b/internal/bits/max_amd64.s
@@ -458,7 +458,7 @@ TEXT ·maxBE128(SB), NOSPLIT, $-48
     CMPQ DX, $256
     JB loop
 
-    CMPQ ·hasAVX512MinMaxBE128(SB), $0
+    CMPB ·hasAVX512MinMaxBE128(SB), $0
     JB loop
 
     // Z19 holds a vector of the count by which we increment the vectors of
diff --git a/internal/bits/min_amd64.s b/internal/bits/min_amd64.s
index 12bb461..07ec94c 100644
--- a/internal/bits/min_amd64.s
+++ b/internal/bits/min_amd64.s
@@ -459,7 +459,7 @@ TEXT ·minBE128(SB), NOSPLIT, $-48
     CMPQ DX, $256
     JB loop
 
-    CMPQ ·hasAVX512MinMaxBE128(SB), $0
+    CMPB ·hasAVX512MinMaxBE128(SB), $0
     JB loop
 
     // Z19 holds a vector of the count by which we increment the vectors of
diff --git a/page.go b/page.go
index db2fafd..63280d5 100644
--- a/page.go
+++ b/page.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/encoding/plain"
 	"github.com/segmentio/parquet-go/internal/bits"
 )
 
@@ -642,6 +643,19 @@ type booleanPageReader struct {
 	offset int
 }
 
+func (r *booleanPageReader) Read(b []byte) (n int, err error) {
+	return r.ReadBooleans(bits.BytesToBool(b))
+}
+
+func (r *booleanPageReader) ReadBooleans(values []bool) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *booleanPageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueBoolean(r.page.values[r.offset])
@@ -716,6 +730,20 @@ type int32PageReader struct {
 	offset int
 }
 
+func (r *int32PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt32s(bits.BytesToInt32(b))
+	return 4 * n, err
+}
+
+func (r *int32PageReader) ReadInt32s(values []int32) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *int32PageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueInt32(r.page.values[r.offset])
@@ -790,6 +818,20 @@ type int64PageReader struct {
 	offset int
 }
 
+func (r *int64PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt64s(bits.BytesToInt64(b))
+	return 8 * n, err
+}
+
+func (r *int64PageReader) ReadInt64s(values []int64) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *int64PageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueInt64(r.page.values[r.offset])
@@ -866,6 +908,20 @@ type int96PageReader struct {
 	offset int
 }
 
+func (r *int96PageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadInt96s(deprecated.BytesToInt96(b))
+	return 12 * n, err
+}
+
+func (r *int96PageReader) ReadInt96s(values []deprecated.Int96) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *int96PageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueInt96(r.page.values[r.offset])
@@ -940,6 +996,20 @@ type floatPageReader struct {
 	offset int
 }
 
+func (r *floatPageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadFloats(bits.BytesToFloat32(b))
+	return 4 * n, err
+}
+
+func (r *floatPageReader) ReadFloats(values []float32) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *floatPageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueFloat(r.page.values[r.offset])
@@ -1014,6 +1084,20 @@ type doublePageReader struct {
 	offset int
 }
 
+func (r *doublePageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadDoubles(bits.BytesToFloat64(b))
+	return 8 * n, err
+}
+
+func (r *doublePageReader) ReadDoubles(values []float64) (n int, err error) {
+	n = copy(values, r.page.values[r.offset:])
+	r.offset += n
+	if r.offset == len(r.page.values) {
+		err = io.EOF
+	}
+	return n, err
+}
+
 func (r *doublePageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		values[n] = makeValueDouble(r.page.values[r.offset])
@@ -1126,6 +1210,37 @@ type byteArrayPageReader struct {
 	offset int
 }
 
+func (r *byteArrayPageReader) Read(b []byte) (int, error) {
+	_, n, err := r.readByteArrays(b)
+	return n, err
+}
+
+func (r *byteArrayPageReader) ReadByteArrays(values []byte) (int, error) {
+	n, _, err := r.readByteArrays(values)
+	return n, err
+}
+
+func (r *byteArrayPageReader) readByteArrays(values []byte) (c, n int, err error) {
+	for r.offset < r.page.values.Len() {
+		b := r.page.values.Index(r.offset)
+		k := plain.ByteArrayLengthSize + len(b)
+		if k > (len(values) - n) {
+			break
+		}
+		plain.PutByteArrayLength(values[n:], len(b))
+		n += plain.ByteArrayLengthSize
+		n += copy(values[n:], b)
+		r.offset++
+		c++
+	}
+	if r.offset == r.page.values.Len() {
+		err = io.EOF
+	} else if n == 0 && len(values) > 0 {
+		err = io.ErrShortBuffer
+	}
+	return c, n, err
+}
+
 func (r *byteArrayPageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < r.page.values.Len() {
 		values[n] = makeValueBytes(ByteArray, r.page.values.Index(r.offset))
@@ -1213,6 +1328,22 @@ type fixedLenByteArrayPageReader struct {
 	offset int
 }
 
+func (r *fixedLenByteArrayPageReader) Read(b []byte) (n int, err error) {
+	n, err = r.ReadFixedLenByteArrays(b)
+	return n * r.page.size, err
+}
+
+func (r *fixedLenByteArrayPageReader) ReadFixedLenByteArrays(values []byte) (n int, err error) {
+	n = copy(values, r.page.data[r.offset:]) / r.page.size
+	r.offset += n * r.page.size
+	if r.offset == len(r.page.data) {
+		err = io.EOF
+	} else if n == 0 && len(values) > 0 {
+		err = io.ErrShortBuffer
+	}
+	return n, err
+}
+
 func (r *fixedLenByteArrayPageReader) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.data) {
 		values[n] = makeValueBytes(FixedLenByteArray, r.page.data[r.offset:r.offset+r.page.size])
@@ -1357,4 +1488,22 @@ var (
 
 	_ io.ReaderFrom = (*errorBuffer)(nil)
 	_ io.WriterTo   = (*errorBuffer)(nil)
+
+	_ io.Reader = (*booleanPageReader)(nil)
+	_ io.Reader = (*int32PageReader)(nil)
+	_ io.Reader = (*int64PageReader)(nil)
+	_ io.Reader = (*int96PageReader)(nil)
+	_ io.Reader = (*floatPageReader)(nil)
+	_ io.Reader = (*doublePageReader)(nil)
+	_ io.Reader = (*byteArrayPageReader)(nil)
+	_ io.Reader = (*fixedLenByteArrayPageReader)(nil)
+
+	_ BooleanReader           = (*booleanPageReader)(nil)
+	_ Int32Reader             = (*int32PageReader)(nil)
+	_ Int64Reader             = (*int64PageReader)(nil)
+	_ Int96Reader             = (*int96PageReader)(nil)
+	_ FloatReader             = (*floatPageReader)(nil)
+	_ DoubleReader            = (*doublePageReader)(nil)
+	_ ByteArrayReader         = (*byteArrayPageReader)(nil)
+	_ FixedLenByteArrayReader = (*fixedLenByteArrayPageReader)(nil)
 )
diff --git a/row.go b/row.go
index 75ba2e7..d814bf4 100644
--- a/row.go
+++ b/row.go
@@ -54,7 +54,7 @@ type RowReader interface {
 
 // RowReaderAt reads parquet rows at specific indexes.
 type RowReaderAt interface {
-	ReadRowAt(Row, int) (Row, error)
+	ReadRowAt(Row, int64) (Row, error)
 }
 
 // RowReaderFrom reads parquet rows from reader.
@@ -83,7 +83,7 @@ type RowWriter interface {
 
 // RowWriterAt writes parquet rows at specific indexes.
 type RowWriterAt interface {
-	WriteRowAt(Row, int) error
+	WriteRowAt(Row, int64) error
 }
 
 // RowWriterTo writes parquet rows to a writer.
@@ -214,15 +214,15 @@ func targetSchemaOf(w RowWriter) *Schema {
 	return nil
 }
 
-func errRowIndexOutOfBounds(rowIndex, rowCount int) error {
+func errRowIndexOutOfBounds(rowIndex, rowCount int64) error {
 	return fmt.Errorf("row index out of bounds: %d/%d", rowIndex, rowCount)
 }
 
-func errRowHasTooFewValues(numValues int) error {
+func errRowHasTooFewValues(numValues int64) error {
 	return fmt.Errorf("row has too few values to be written to the column: %d", numValues)
 }
 
-func errRowHasTooManyValues(numValues int) error {
+func errRowHasTooManyValues(numValues int64) error {
 	return fmt.Errorf("row has too many values to be written to the column: %d", numValues)
 }
 
diff --git a/value.go b/value.go
index 3192187..2a1323d 100644
--- a/value.go
+++ b/value.go
@@ -120,6 +120,176 @@ func copyValues(dst ValueWriter, src ValueReader, buf []Value) (written int64, e
 	}
 }
 
+// BooleanReader is an interface implemented by ValueReader instances which
+// expose the content of a column of boolean values.
+type BooleanReader interface {
+	// Read boolean values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadBooleans(values []bool) (int, error)
+}
+
+// BooleanWriter is an interface implemented by ValueWriter instances which
+// support writing columns of boolean values.
+type BooleanWriter interface {
+	// Write boolean values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteBooleans(values []bool) (int, error)
+}
+
+// Int32Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int32 values.
+type Int32Reader interface {
+	// Read 32 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt32s(values []int32) (int, error)
+}
+
+// Int32Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 32 bits signed integer values.
+type Int32Writer interface {
+	// Write 32 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt32s(values []int32) (int, error)
+}
+
+// Int64Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int64 values.
+type Int64Reader interface {
+	// Read 64 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt64s(values []int64) (int, error)
+}
+
+// Int64Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 64 bits signed integer values.
+type Int64Writer interface {
+	// Write 64 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt64s(values []int64) (int, error)
+}
+
+// Int96Reader is an interface implemented by ValueReader instances which expose
+// the content of a column of int96 values.
+type Int96Reader interface {
+	// Read 96 bits integer values into the buffer passed as argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadInt96s(values []deprecated.Int96) (int, error)
+}
+
+// Int96Writer is an interface implemented by ValueWriter instances which
+// support writing columns of 96 bits signed integer values.
+type Int96Writer interface {
+	// Write 96 bits signed integer values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteInt96s(values []deprecated.Int96) (int, error)
+}
+
+// FloatReader is an interface implemented by ValueReader instances which expose
+// the content of a column of single-precision floating point values.
+type FloatReader interface {
+	// Read single-precision floating point values into the buffer passed as
+	// argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadFloats(values []float32) (int, error)
+}
+
+// FloatWriter is an interface implemented by ValueWriter instances which
+// support writing columns of single-precision floating point values.
+type FloatWriter interface {
+	// Write single-precision floating point values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteFloats(values []float32) (int, error)
+}
+
+// DoubleReader is an interface implemented by ValueReader instances which
+// expose the content of a column of double-precision float point values.
+type DoubleReader interface {
+	// Read double-precision floating point values into the buffer passed as
+	// argument.
+	//
+	// The method returns io.EOF when all values have been read.
+	ReadDoubles(values []float64) (int, error)
+}
+
+// DoubleWriter is an interface implemented by ValueWriter instances which
+// support writing columns of double-precision floating point values.
+type DoubleWriter interface {
+	// Write double-precision floating point values.
+	//
+	// The method returns the number of values written, and any error that
+	// occured while writing the values.
+	WriteDoubles(values []float64) (int, error)
+}
+
+// ByteArrayReader is an interface implemented by ValueReader instances which
+// expose the content of a column of variable length byte array values.
+type ByteArrayReader interface {
+	// Read values into the byte buffer passed as argument, returning the number
+	// of values written to the buffer (not the number of bytes). Values are
+	// written using the PLAIN encoding, each byte array prefixed with its
+	// length encoded as a 4 bytes little endian unsigned integer.
+	//
+	// The method returns io.EOF when all values have been read.
+	//
+	// If the buffer was not empty, but too small to hold at least one value,
+	// io.ErrShortBuffer is returned.
+	ReadByteArrays(values []byte) (int, error)
+}
+
+// ByteArrayWriter is an interface implemented by ValueWriter instances which
+// support writing columns of variable length byte array values.
+type ByteArrayWriter interface {
+	// Write variable length byte array values.
+	//
+	// The values passed as input must be laid out using the PLAIN encoding,
+	// with each byte array prefixed with the four bytes little endian unsigned
+	// integer length.
+	//
+	// The method returns the number of values written to the underlying column
+	// (not the number of bytes), or any error that occured while attempting to
+	// write the values.
+	WriteByteArrays(values []byte) (int, error)
+}
+
+// FixedLenByteArrayReader is an interface implemented by ValueReader instances
+// which expose the content of a column of fixed length byte array values.
+type FixedLenByteArrayReader interface {
+	// Read values into the byte buffer passed as argument, returning the number
+	// of values written to the buffer (not the number of bytes).
+	//
+	// The method returns io.EOF when all values have been read.
+	//
+	// If the buffer was not empty, but too small to hold at least one value,
+	// io.ErrShortBuffer is returned.
+	ReadFixedLenByteArrays(values []byte) (int, error)
+}
+
+// FixedLenByteArrayWriter is an interface implemented by ValueWriter instances
+// which support writing columns of fixed length byte array values.
+type FixedLenByteArrayWriter interface {
+	// Writes the fixed length byte array values.
+	//
+	// The size of the values is assumed to be the same as the expected size of
+	// items in the column. The method errors if the length of the input values
+	// is not a multiple of the expected item size.
+	WriteFixedLenByteArrays(values []byte) (int, error)
+}
+
 // ValueOf constructs a parquet value from a Go value v.
 //
 // The physical type of the value is assumed from the Go type of v using the


Test patch:
diff --git a/page_test.go b/page_test.go
new file mode 100644
index 0000000..036ceff
--- /dev/null
+++ b/page_test.go
@@ -0,0 +1,341 @@
+package parquet_test
+
+import (
+	"io"
+	"reflect"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding/plain"
+	"github.com/segmentio/parquet-go/internal/bits"
+)
+
+func TestPage(t *testing.T) {
+	t.Run("BOOLEAN", testPageBoolean)
+	t.Run("INT32", testPageInt32)
+	t.Run("INT64", testPageInt64)
+	t.Run("INT96", testPageInt96)
+	t.Run("FLOAT", testPageFloat)
+	t.Run("DOUBLE", testPageDouble)
+	t.Run("BYTE_ARRAY", testPageByteArray)
+	t.Run("FIXED_LEN_BYTE_ARRAY", testPageFixedLenByteArray)
+}
+
+func testPageBoolean(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value bool }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []bool{false, true}
+				n, err := w.(io.Writer).Write(bits.BoolToBytes(values))
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]bool, 2)
+				n, err := r.(io.Reader).Read(bits.BoolToBytes(values))
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []bool{false, true}
+				n, err := w.(parquet.BooleanWriter).WriteBooleans(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]bool, 2)
+				n, err := r.(parquet.BooleanReader).ReadBooleans(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageInt32(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value int32 }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(io.Writer).Write(bits.Int32ToBytes(values))
+				return values[:n/4], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]int32, 10)
+				n, err := r.(io.Reader).Read(bits.Int32ToBytes(values))
+				return values[:n/4], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []int32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(parquet.Int32Writer).WriteInt32s(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]int32, 10)
+				n, err := r.(parquet.Int32Reader).ReadInt32s(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageInt64(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value int64 }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(io.Writer).Write(bits.Int64ToBytes(values))
+				return values[:n/8], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]int64, 10)
+				n, err := r.(io.Reader).Read(bits.Int64ToBytes(values))
+				return values[:n/8], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(parquet.Int64Writer).WriteInt64s(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]int64, 10)
+				n, err := r.(parquet.Int64Reader).ReadInt64s(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageInt96(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value deprecated.Int96 }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []deprecated.Int96{{0: 0}, {0: 1}, {0: 2}}
+				n, err := w.(io.Writer).Write(deprecated.Int96ToBytes(values))
+				return values[:n/12], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]deprecated.Int96, 3)
+				n, err := r.(io.Reader).Read(deprecated.Int96ToBytes(values))
+				return values[:n/12], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []deprecated.Int96{{0: 0}, {0: 1}, {0: 2}}
+				n, err := w.(parquet.Int96Writer).WriteInt96s(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]deprecated.Int96, 3)
+				n, err := r.(parquet.Int96Reader).ReadInt96s(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageFloat(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value float32 }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []float32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(io.Writer).Write(bits.Float32ToBytes(values))
+				return values[:n/4], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]float32, 10)
+				n, err := r.(io.Reader).Read(bits.Float32ToBytes(values))
+				return values[:n/4], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []float32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(parquet.FloatWriter).WriteFloats(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]float32, 10)
+				n, err := r.(parquet.FloatReader).ReadFloats(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageDouble(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value float64 }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []float64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(io.Writer).Write(bits.Float64ToBytes(values))
+				return values[:n/8], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]float64, 10)
+				n, err := r.(io.Reader).Read(bits.Float64ToBytes(values))
+				return values[:n/8], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []float64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+				n, err := w.(parquet.DoubleWriter).WriteDoubles(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]float64, 10)
+				n, err := r.(parquet.DoubleReader).ReadDoubles(values)
+				return values[:n], err
+			},
+		})
+	})
+}
+
+func testPageByteArray(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value []byte }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []byte{}
+				values = plain.AppendByteArray(values, []byte("A"))
+				values = plain.AppendByteArray(values, []byte("B"))
+				values = plain.AppendByteArray(values, []byte("C"))
+				n, err := w.(io.Writer).Write(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]byte, 3+3*plain.ByteArrayLengthSize)
+				n, err := r.(io.Reader).Read(values)
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []byte{}
+				values = plain.AppendByteArray(values, []byte("A"))
+				values = plain.AppendByteArray(values, []byte("B"))
+				values = plain.AppendByteArray(values, []byte("C"))
+				_, err := w.(parquet.ByteArrayWriter).WriteByteArrays(values)
+				return values, err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]byte, 3+3*plain.ByteArrayLengthSize)
+				_, err := r.(parquet.ByteArrayReader).ReadByteArrays(values)
+				return values, err
+			},
+		})
+	})
+}
+
+func testPageFixedLenByteArray(t *testing.T) {
+	schema := parquet.SchemaOf(struct{ Value [3]byte }{})
+
+	t.Run("io", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []byte("123456789")
+				n, err := w.(io.Writer).Write(values)
+				return values[:n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]byte, 3*3)
+				n, err := r.(io.Reader).Read(values)
+				return values[:n], err
+			},
+		})
+	})
+
+	t.Run("parquet", func(t *testing.T) {
+		testPage(t, schema, pageTest{
+			write: func(w parquet.ValueWriter) (interface{}, error) {
+				values := []byte("123456789")
+				n, err := w.(parquet.FixedLenByteArrayWriter).WriteFixedLenByteArrays(values)
+				return values[:3*n], err
+			},
+
+			read: func(r parquet.ValueReader) (interface{}, error) {
+				values := make([]byte, 3*3)
+				n, err := r.(parquet.FixedLenByteArrayReader).ReadFixedLenByteArrays(values)
+				return values[:3*n], err
+			},
+		})
+	})
+}
+
+type pageTest struct {
+	write func(parquet.ValueWriter) (interface{}, error)
+	read  func(parquet.ValueReader) (interface{}, error)
+}
+
+func testPage(t *testing.T, schema *parquet.Schema, test pageTest) {
+	buffer := parquet.NewBuffer(schema)
+	column := buffer.Column(0).(parquet.ColumnBuffer)
+
+	w, err := test.write(column)
+	if err != nil {
+		t.Fatal("writing page values:", err)
+	}
+
+	r, err := test.read(column.Page().Values())
+	if err != io.EOF {
+		t.Errorf("expected io.EOF after reading all values but got %v", err)
+	}
+	if !reflect.DeepEqual(w, r) {
+		t.Errorf("wrong values read from the page: got=%+v want=%+v", r, w)
+	}
+
+}