Repository: segmentio/parquet-go
Pull request #426 (2022-12-05 06:40:27)

add parquet.RowBuffer
This PR adds a new type named `parquet.RowBuffer` which is similar to `parquet.GenericBuffer` but uses a row layout to hold data in memory (instead of a column layout).

I left documentation in the code, please take a look at let me know if anything should be changed.


Patch:
diff --git a/bloom.go b/bloom.go
index f84abf2..4ea94ff 100644
--- a/bloom.go
+++ b/bloom.go
@@ -44,15 +44,13 @@ func (f *bloomFilter) Check(v Value) (bool, error) {
 func (v Value) hash(h bloom.Hash) uint64 {
 	switch v.Kind() {
 	case Boolean:
-		return h.Sum64Uint8(uint8(v.u64))
+		return h.Sum64Uint8(v.byte())
 	case Int32, Float:
-		return h.Sum64Uint32(uint32(v.u64))
+		return h.Sum64Uint32(v.uint32())
 	case Int64, Double:
-		return h.Sum64Uint64(v.u64)
-	case Int96:
-		return h.Sum64(v.Bytes())
-	default:
-		return h.Sum64(v.ByteArray())
+		return h.Sum64Uint64(v.uint64())
+	default: // Int96, ByteArray, FixedLenByteArray, or null
+		return h.Sum64(v.byteArray())
 	}
 }
 
diff --git a/column_buffer.go b/column_buffer.go
index d144b79..038eba4 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -1602,7 +1602,7 @@ func (col *fixedLenByteArrayColumnBuffer) WriteFixedLenByteArrays(values []byte)
 
 func (col *fixedLenByteArrayColumnBuffer) WriteValues(values []Value) (int, error) {
 	for _, v := range values {
-		col.data = append(col.data, v.ByteArray()...)
+		col.data = append(col.data, v.byteArray()...)
 	}
 	return len(values), nil
 }
@@ -1893,7 +1893,7 @@ func (col *be128ColumnBuffer) WriteValues(values []Value) (int, error) {
 	col.values = col.values[:n+len(values)]
 	newValues := col.values[n:]
 	for i, v := range values {
-		copy(newValues[i][:], v.ByteArray())
+		copy(newValues[i][:], v.byteArray())
 	}
 	return len(values), nil
 }
diff --git a/column_index.go b/column_index.go
index 8df2702..01b2255 100644
--- a/column_index.go
+++ b/column_index.go
@@ -315,8 +315,8 @@ func (i *booleanColumnIndexer) Reset() {
 
 func (i *booleanColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Boolean())
-	i.maxValues = append(i.maxValues, max.Boolean())
+	i.minValues = append(i.minValues, min.boolean())
+	i.maxValues = append(i.maxValues, max.boolean())
 }
 
 func (i *booleanColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -346,8 +346,8 @@ func (i *int32ColumnIndexer) Reset() {
 
 func (i *int32ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Int32())
-	i.maxValues = append(i.maxValues, max.Int32())
+	i.minValues = append(i.minValues, min.int32())
+	i.maxValues = append(i.maxValues, max.int32())
 }
 
 func (i *int32ColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -377,8 +377,8 @@ func (i *int64ColumnIndexer) Reset() {
 
 func (i *int64ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Int64())
-	i.maxValues = append(i.maxValues, max.Int64())
+	i.minValues = append(i.minValues, min.int64())
+	i.maxValues = append(i.maxValues, max.int64())
 }
 
 func (i *int64ColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -439,8 +439,8 @@ func (i *floatColumnIndexer) Reset() {
 
 func (i *floatColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Float())
-	i.maxValues = append(i.maxValues, max.Float())
+	i.minValues = append(i.minValues, min.float())
+	i.maxValues = append(i.maxValues, max.float())
 }
 
 func (i *floatColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -470,8 +470,8 @@ func (i *doubleColumnIndexer) Reset() {
 
 func (i *doubleColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Double())
-	i.maxValues = append(i.maxValues, max.Double())
+	i.minValues = append(i.minValues, min.double())
+	i.maxValues = append(i.maxValues, max.double())
 }
 
 func (i *doubleColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -502,8 +502,8 @@ func (i *byteArrayColumnIndexer) Reset() {
 
 func (i *byteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	minValue := min.ByteArray()
-	maxValue := max.ByteArray()
+	minValue := min.byteArray()
+	maxValue := max.byteArray()
 	if i.sizeLimit > 0 {
 		minValue = truncateLargeMinByteArrayValue(minValue, i.sizeLimit)
 		maxValue = truncateLargeMaxByteArrayValue(maxValue, i.sizeLimit)
@@ -546,8 +546,8 @@ func (i *fixedLenByteArrayColumnIndexer) Reset() {
 
 func (i *fixedLenByteArrayColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.ByteArray()...)
-	i.maxValues = append(i.maxValues, max.ByteArray()...)
+	i.minValues = append(i.minValues, min.byteArray()...)
+	i.maxValues = append(i.maxValues, max.byteArray()...)
 }
 
 func (i *fixedLenByteArrayColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -587,8 +587,8 @@ func (i *uint32ColumnIndexer) Reset() {
 
 func (i *uint32ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Uint32())
-	i.maxValues = append(i.maxValues, max.Uint32())
+	i.minValues = append(i.minValues, min.uint32())
+	i.maxValues = append(i.maxValues, max.uint32())
 }
 
 func (i *uint32ColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -618,8 +618,8 @@ func (i *uint64ColumnIndexer) Reset() {
 
 func (i *uint64ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
-	i.minValues = append(i.minValues, min.Uint64())
-	i.maxValues = append(i.maxValues, max.Uint64())
+	i.minValues = append(i.minValues, min.uint64())
+	i.maxValues = append(i.maxValues, max.uint64())
 }
 
 func (i *uint64ColumnIndexer) ColumnIndex() format.ColumnIndex {
@@ -650,10 +650,10 @@ func (i *be128ColumnIndexer) Reset() {
 func (i *be128ColumnIndexer) IndexPage(numValues, numNulls int64, min, max Value) {
 	i.observe(numValues, numNulls)
 	if !min.IsNull() {
-		i.minValues = append(i.minValues, *(*[16]byte)(min.ByteArray()))
+		i.minValues = append(i.minValues, *(*[16]byte)(min.byteArray()))
 	}
 	if !max.IsNull() {
-		i.maxValues = append(i.maxValues, *(*[16]byte)(max.ByteArray()))
+		i.maxValues = append(i.maxValues, *(*[16]byte)(max.byteArray()))
 	}
 }
 
diff --git a/compare.go b/compare.go
index f0d55dd..8f6f33c 100644
--- a/compare.go
+++ b/compare.go
@@ -6,6 +6,54 @@ import (
 	"github.com/segmentio/parquet-go/deprecated"
 )
 
+// CompareDescending constructs a comparison function which inverses the order
+// of values.
+//
+//go:noinline
+func CompareDescending(cmp func(Value, Value) int) func(Value, Value) int {
+	return func(a, b Value) int { return -cmp(a, b) }
+}
+
+// CompareNullsFirst constructs a comparison function which assumes that null
+// values are smaller than all other values.
+//
+//go:noinline
+func CompareNullsFirst(cmp func(Value, Value) int) func(Value, Value) int {
+	return func(a, b Value) int {
+		switch {
+		case a.IsNull():
+			if b.IsNull() {
+				return 0
+			}
+			return -1
+		case b.IsNull():
+			return +1
+		default:
+			return cmp(a, b)
+		}
+	}
+}
+
+// CompareNullsLast constructs a comparison function which assumes that null
+// values are greater than all other values.
+//
+//go:noinline
+func CompareNullsLast(cmp func(Value, Value) int) func(Value, Value) int {
+	return func(a, b Value) int {
+		switch {
+		case a.IsNull():
+			if b.IsNull() {
+				return 0
+			}
+			return +1
+		case b.IsNull():
+			return -1
+		default:
+			return cmp(a, b)
+		}
+	}
+}
+
 func compareBool(v1, v2 bool) int {
 	switch {
 	case !v1 && v2:
@@ -128,3 +176,135 @@ func lessBE128(v1, v2 *[16]byte) bool {
 	y = binary.BigEndian.Uint64(v2[8:])
 	return x < y
 }
+
+func compareRowsFuncOf(schema *Schema, sortingColumns []SortingColumn) func(Row, Row) int {
+	compareFuncs := make([]func(Row, Row) int, 0, len(sortingColumns))
+	direct := true
+
+	for _, column := range schema.Columns() {
+		leaf, _ := schema.Lookup(column...)
+		if leaf.MaxRepetitionLevel > 0 {
+			direct = false
+		}
+
+		for _, sortingColumn := range sortingColumns {
+			path1 := columnPath(column)
+			path2 := columnPath(sortingColumn.Path())
+
+			if path1.equal(path2) {
+				descending := sortingColumn.Descending()
+				optional := leaf.MaxDefinitionLevel > 0
+				sortFunc := (func(Row, Row) int)(nil)
+
+				if direct && !optional {
+					// This is an optimization for the common case where rows
+					// are sorted by non-optional, non-repeated columns.
+					//
+					// The sort function can make the assumption that it will
+					// find the column value at the current column index, and
+					// does not need to scan the rows looking for values with
+					// a matching column index.
+					//
+					// A second optimization consists in passing the column type
+					// directly to the sort function instead of an intermediary
+					// closure, which removes an indirection layer and improves
+					// throughput by ~20% in BenchmarkSortRowBuffer.
+					typ := leaf.Node.Type()
+					if descending {
+						sortFunc = compareRowsFuncOfIndexDescending(leaf.ColumnIndex, typ)
+					} else {
+						sortFunc = compareRowsFuncOfIndexAscending(leaf.ColumnIndex, typ)
+					}
+				} else {
+					compare := leaf.Node.Type().Compare
+
+					if descending {
+						compare = CompareDescending(compare)
+					}
+
+					if optional {
+						if sortingColumn.NullsFirst() {
+							compare = CompareNullsFirst(compare)
+						} else {
+							compare = CompareNullsLast(compare)
+						}
+					}
+
+					sortFunc = compareRowsFuncOfScan(leaf.ColumnIndex, compare)
+				}
+
+				compareFuncs = append(compareFuncs, sortFunc)
+			}
+		}
+	}
+
+	// For the common case where rows are sorted by a single column, we can skip
+	// looping over the list of sort functions.
+	switch len(compareFuncs) {
+	case 0:
+		return compareRowsUnordered
+	case 1:
+		return compareFuncs[0]
+	default:
+		return compareRowsFuncOfColumns(compareFuncs)
+	}
+}
+
+func compareRowsUnordered(Row, Row) int { return 0 }
+
+//go:noinline
+func compareRowsFuncOfColumns(compareFuncs []func(Row, Row) int) func(Row, Row) int {
+	return func(row1, row2 Row) int {
+		for _, compare := range compareFuncs {
+			if cmp := compare(row1, row2); cmp != 0 {
+				return cmp
+			}
+		}
+		return 0
+	}
+}
+
+//go:noinline
+func compareRowsFuncOfIndexAscending(columnIndex int, typ Type) func(Row, Row) int {
+	return func(row1, row2 Row) int { return typ.Compare(row1[columnIndex], row2[columnIndex]) }
+}
+
+//go:noinline
+func compareRowsFuncOfIndexDescending(columnIndex int, typ Type) func(Row, Row) int {
+	return func(row1, row2 Row) int { return -typ.Compare(row1[columnIndex], row2[columnIndex]) }
+}
+
+//go:noinline
+func compareRowsFuncOfScan(columnIndex int, compare func(Value, Value) int) func(Row, Row) int {
+	columnIndexOfValues := ^int16(columnIndex)
+	return func(row1, row2 Row) int {
+		i1 := 0
+		i2 := 0
+
+		for {
+			for i1 < len(row1) && row1[i1].columnIndex != columnIndexOfValues {
+				i1++
+			}
+
+			for i2 < len(row2) && row2[i2].columnIndex != columnIndexOfValues {
+				i2++
+			}
+
+			end1 := i1 == len(row1)
+			end2 := i2 == len(row2)
+
+			if end1 && end2 {
+				return 0
+			} else if end1 {
+				return -1
+			} else if end2 {
+				return +1
+			} else if cmp := compare(row1[i1], row2[i2]); cmp != 0 {
+				return cmp
+			}
+
+			i1++
+			i2++
+		}
+	}
+}
diff --git a/internal/unsafecast/unsafecast_go17.go b/internal/unsafecast/unsafecast_go17.go
index 82113a6..8e7b16b 100644
--- a/internal/unsafecast/unsafecast_go17.go
+++ b/internal/unsafecast/unsafecast_go17.go
@@ -134,3 +134,7 @@ func BytesToFloat64(data []byte) []float64 {
 func BytesToString(data []byte) string {
 	return *(*string)(unsafe.Pointer(&data))
 }
+
+func Bytes(data *byte, size int) []byte {
+	return unsafe.Slice(data, size)
+}
diff --git a/internal/unsafecast/unsafecast_go18.go b/internal/unsafecast/unsafecast_go18.go
index a30eb6a..e3b2ca7 100644
--- a/internal/unsafecast/unsafecast_go18.go
+++ b/internal/unsafecast/unsafecast_go18.go
@@ -81,6 +81,16 @@ func Slice[To, From any](data []From) []To {
 	return *(*[]To)(unsafe.Pointer(s))
 }
 
+// Bytes constructs a byte slice. The pointer to the first element of the slice
+// is set to data, the length and capacity are set to size.
+func Bytes(data *byte, size int) []byte {
+	return *(*[]byte)(unsafe.Pointer(&slice{
+		ptr: unsafe.Pointer(data),
+		len: size,
+		cap: size,
+	}))
+}
+
 // BytesToString converts a byte slice to a string value. The returned string
 // shares the backing array of the byte slice.
 //
diff --git a/row_buffer.go b/row_buffer.go
new file mode 100644
index 0000000..0d45778
--- /dev/null
+++ b/row_buffer.go
@@ -0,0 +1,505 @@
+//go:build go1.18
+
+package parquet
+
+import (
+	"io"
+	"sort"
+
+	"github.com/segmentio/parquet-go/deprecated"
+	"github.com/segmentio/parquet-go/encoding"
+	"github.com/segmentio/parquet-go/internal/unsafecast"
+)
+
+// RowBuffer is an implementation of the RowGroup interface which stores parquet
+// rows in memory.
+//
+// Unlike GenericBuffer which uses a column layout to store values in memory
+// buffers, RowBuffer uses a row layout. The use of row layout provides greater
+// efficiency when sorting the buffer, which is the primary use case for the
+// RowBuffer type. Applications which intend to sort rows prior to writing them
+// to a parquet file will often see lower CPU utilization from using a RowBuffer
+// than a GenericBuffer.
+//
+// RowBuffer values are not safe to use concurrently from multiple goroutines.
+type RowBuffer[T any] struct {
+	schema  *Schema
+	sorting []SortingColumn
+	buffer  []byte
+	rows    []Row
+	numRows int
+	compare func(Row, Row) int
+}
+
+// NewRowBuffer constructs a new row buffer.
+func NewRowBuffer[T any](options ...RowGroupOption) *RowBuffer[T] {
+	config := DefaultRowGroupConfig()
+	config.Apply(options...)
+	if err := config.Validate(); err != nil {
+		panic(err)
+	}
+
+	t := typeOf[T]()
+	if config.Schema == nil && t != nil {
+		config.Schema = schemaOf(dereference(t))
+	}
+
+	if config.Schema == nil {
+		panic("row buffer must be instantiated with schema or concrete type.")
+	}
+
+	return &RowBuffer[T]{
+		schema:  config.Schema,
+		sorting: config.SortingColumns,
+		compare: config.Schema.Comparator(config.SortingColumns...),
+	}
+}
+
+// Reset clears the content of the buffer without releasing its memory.
+func (buf *RowBuffer[T]) Reset() {
+	for _, row := range buf.rows[:buf.numRows] {
+		for i := range row {
+			row[i] = Value{}
+		}
+	}
+	buf.buffer = buf.buffer[:0]
+	buf.numRows = 0
+}
+
+// NumRows returns the number of rows currently written to the buffer.
+func (buf *RowBuffer[T]) NumRows() int64 { return int64(buf.numRows) }
+
+// ColumnChunks returns a view of the buffer's columns.
+//
+// Note that reading columns of a RowBuffer will be less efficient than reading
+// columns of a GenericBuffer since the latter uses a column layout. This method
+// is mainly exposed to satisfy the RowGroup interface, applications which need
+// compute-efficient column scans on in-memory buffers should likely use a
+// GenericBuffer instead.
+//
+// The returned column chunks are snapshots at the time the method is called,
+// they remain valid until the next call to Reset on the buffer.
+func (buf *RowBuffer[T]) ColumnChunks() []ColumnChunk {
+	columns := buf.schema.Columns()
+	chunks := make([]rowBufferColumnChunk, len(columns))
+
+	for i, column := range columns {
+		leafColumn, _ := buf.schema.Lookup(column...)
+		chunks[i] = rowBufferColumnChunk{
+			page: rowBufferPage{
+				rows:               buf.rows[:buf.numRows],
+				typ:                leafColumn.Node.Type(),
+				column:             leafColumn.ColumnIndex,
+				maxRepetitionLevel: byte(leafColumn.MaxRepetitionLevel),
+				maxDefinitionLevel: byte(leafColumn.MaxDefinitionLevel),
+			},
+		}
+	}
+
+	columnChunks := make([]ColumnChunk, len(chunks))
+	for i := range chunks {
+		columnChunks[i] = &chunks[i]
+	}
+	return columnChunks
+}
+
+// SortingColumns returns the list of columns that rows are expected to be
+// sorted by.
+//
+// The list of sorting columns is configured when the buffer is created and used
+// when it is sorted.
+//
+// Note that unless the buffer is explicitly sorted, there are no guarantees
+// that the rows it contains will be in the order specified by the sorting
+// columns.
+func (buf *RowBuffer[T]) SortingColumns() []SortingColumn { return buf.sorting }
+
+// Schema returns the schema of rows in the buffer.
+func (buf *RowBuffer[T]) Schema() *Schema { return buf.schema }
+
+// Len returns the number of rows in the buffer.
+//
+// The method contributes to satisfying sort.Interface.
+func (buf *RowBuffer[T]) Len() int { return buf.numRows }
+
+// Less compares the rows at index i and j according to the sorting columns
+// configured on the buffer.
+//
+// The method contributes to satisfying sort.Interface.
+func (buf *RowBuffer[T]) Less(i, j int) bool {
+	return buf.compare(buf.rows[i], buf.rows[j]) < 0
+}
+
+// Swap exchanges the rows at index i and j in the buffer.
+//
+// The method contributes to satisfying sort.Interface.
+func (buf *RowBuffer[T]) Swap(i, j int) {
+	buf.rows[i], buf.rows[j] = buf.rows[j], buf.rows[i]
+}
+
+// Rows returns a Rows instance exposing rows stored in the buffer.
+//
+// The rows returned are a snapshot at the time the method is called.
+// The returned rows and values read from it remain valid until the next call
+// to Reset on the buffer.
+func (buf *RowBuffer[T]) Rows() Rows {
+	return &rowBufferRows{rows: buf.rows[:buf.numRows], schema: buf.schema}
+}
+
+// Write writes rows to the buffer, returning the number of rows written.
+func (buf *RowBuffer[T]) Write(rows []T) (int, error) {
+	buf.reserve(len(rows))
+
+	for i := range rows {
+		bufRow := buf.rows[buf.numRows][:0]
+		bufRow = buf.schema.Deconstruct(bufRow, &rows[i])
+		buf.capture(bufRow)
+		buf.rows[buf.numRows] = bufRow
+		buf.numRows++
+	}
+
+	return len(rows), nil
+}
+
+// WriteRows writes parquet rows to the buffer, returing the number of rows
+// written.
+func (buf *RowBuffer[T]) WriteRows(rows []Row) (int, error) {
+	buf.reserve(len(rows))
+
+	for _, row := range rows {
+		bufRow := buf.rows[buf.numRows][:0]
+		bufRow = append(bufRow, row...)
+		buf.capture(bufRow)
+		buf.rows[buf.numRows] = bufRow
+		buf.numRows++
+	}
+
+	return len(rows), nil
+}
+
+func (buf *RowBuffer[T]) copyBytes(v []byte) []byte {
+	if free := cap(buf.buffer) - len(buf.buffer); free < len(v) {
+		newCap := 2 * cap(buf.buffer)
+		if newCap == 0 {
+			newCap = defaultReadBufferSize
+		}
+		for newCap < len(v) {
+			newCap *= 2
+		}
+		buf.buffer = make([]byte, 0, newCap)
+	}
+
+	i := len(buf.buffer)
+	j := len(buf.buffer) + len(v)
+	buf.buffer = buf.buffer[:j]
+
+	b := buf.buffer[i:j:j]
+	copy(b, v)
+	return b
+}
+
+func (buf *RowBuffer[T]) capture(row Row) {
+	for i, v := range row {
+		switch kind := v.Kind(); kind {
+		case ByteArray, FixedLenByteArray:
+			row[i].ptr = unsafecast.AddressOfBytes(buf.copyBytes(v.byteArray()))
+		}
+	}
+}
+
+func (buf *RowBuffer[T]) reserve(n int) {
+	if newLen := buf.numRows + n; newLen > len(buf.rows) {
+		bufLen := 2 * len(buf.rows)
+		if bufLen == 0 {
+			bufLen = defaultValueBufferSize
+		}
+		for bufLen < newLen {
+			bufLen *= 2
+		}
+		newRows := make([]Row, bufLen)
+		copy(newRows, buf.rows)
+		buf.rows = newRows
+	}
+}
+
+type rowBufferColumnChunk struct{ page rowBufferPage }
+
+func (c *rowBufferColumnChunk) Type() Type { return c.page.Type() }
+
+func (c *rowBufferColumnChunk) Column() int { return c.page.Column() }
+
+func (c *rowBufferColumnChunk) Pages() Pages { return onePage(&c.page) }
+
+func (c *rowBufferColumnChunk) ColumnIndex() ColumnIndex { return nil }
+
+func (c *rowBufferColumnChunk) OffsetIndex() OffsetIndex { return nil }
+
+func (c *rowBufferColumnChunk) BloomFilter() BloomFilter { return nil }
+
+func (c *rowBufferColumnChunk) NumValues() int64 { return c.page.NumValues() }
+
+type rowBufferPage struct {
+	rows               []Row
+	typ                Type
+	column             int
+	maxRepetitionLevel byte
+	maxDefinitionLevel byte
+}
+
+func (p *rowBufferPage) Type() Type { return p.typ }
+
+func (p *rowBufferPage) Column() int { return p.column }
+
+func (p *rowBufferPage) Dictionary() Dictionary { return nil }
+
+func (p *rowBufferPage) NumRows() int64 { return int64(len(p.rows)) }
+
+func (p *rowBufferPage) NumValues() int64 {
+	numValues := int64(0)
+	p.scan(func(value Value) {
+		if !value.isNull() {
+			numValues++
+		}
+	})
+	return numValues
+}
+
+func (p *rowBufferPage) NumNulls() int64 {
+	numNulls := int64(0)
+	p.scan(func(value Value) {
+		if value.isNull() {
+			numNulls++
+		}
+	})
+	return numNulls
+}
+
+func (p *rowBufferPage) Bounds() (min, max Value, ok bool) {
+	p.scan(func(value Value) {
+		if !value.IsNull() {
+			switch {
+			case !ok:
+				min, max, ok = value, value, true
+			case p.typ.Compare(value, min) < 0:
+				min = value
+			case p.typ.Compare(value, max) > 0:
+				max = value
+			}
+		}
+	})
+	return min, max, ok
+}
+
+func (p *rowBufferPage) Size() int64 { return 0 }
+
+func (p *rowBufferPage) Values() ValueReader {
+	return &rowBufferPageValueReader{
+		page:        p,
+		columnIndex: ^int16(p.column),
+	}
+}
+
+func (p *rowBufferPage) Clone() Page {
+	rows := make([]Row, len(p.rows))
+	for i := range rows {
+		rows[i] = p.rows[i].Clone()
+	}
+	return &rowBufferPage{
+		rows:   rows,
+		typ:    p.typ,
+		column: p.column,
+	}
+}
+
+func (p *rowBufferPage) Slice(i, j int64) Page {
+	return &rowBufferPage{
+		rows:   p.rows[i:j],
+		typ:    p.typ,
+		column: p.column,
+	}
+}
+
+func (p *rowBufferPage) RepetitionLevels() (repetitionLevels []byte) {
+	if p.maxRepetitionLevel != 0 {
+		repetitionLevels = make([]byte, 0, len(p.rows))
+		p.scan(func(value Value) {
+			repetitionLevels = append(repetitionLevels, value.repetitionLevel)
+		})
+	}
+	return repetitionLevels
+}
+
+func (p *rowBufferPage) DefinitionLevels() (definitionLevels []byte) {
+	if p.maxDefinitionLevel != 0 {
+		definitionLevels = make([]byte, 0, len(p.rows))
+		p.scan(func(value Value) {
+			definitionLevels = append(definitionLevels, value.definitionLevel)
+		})
+	}
+	return definitionLevels
+}
+
+func (p *rowBufferPage) Data() encoding.Values {
+	switch p.typ.Kind() {
+	case Boolean:
+		values := make([]byte, (len(p.rows)+7)/8)
+		numValues := 0
+		p.scanNonNull(func(value Value) {
+			if value.boolean() {
+				i := uint(numValues) / 8
+				j := uint(numValues) % 8
+				values[i] |= 1 << j
+			}
+			numValues++
+		})
+		return encoding.BooleanValues(values[:(numValues+7)/8])
+
+	case Int32:
+		values := make([]int32, 0, len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.int32()) })
+		return encoding.Int32Values(values)
+
+	case Int64:
+		values := make([]int64, 0, len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.int64()) })
+		return encoding.Int64Values(values)
+
+	case Int96:
+		values := make([]deprecated.Int96, 0, len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.int96()) })
+		return encoding.Int96Values(values)
+
+	case Float:
+		values := make([]float32, 0, len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.float()) })
+		return encoding.FloatValues(values)
+
+	case Double:
+		values := make([]float64, 0, len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.double()) })
+		return encoding.DoubleValues(values)
+
+	case ByteArray:
+		values := make([]byte, 0, p.typ.EstimateSize(len(p.rows)))
+		offsets := make([]uint32, 0, len(p.rows))
+		p.scanNonNull(func(value Value) {
+			offsets = append(offsets, uint32(len(values)))
+			values = append(values, value.byteArray()...)
+		})
+		offsets = append(offsets, uint32(len(values)))
+		return encoding.ByteArrayValues(values, offsets)
+
+	case FixedLenByteArray:
+		length := p.typ.Length()
+		values := make([]byte, 0, length*len(p.rows))
+		p.scanNonNull(func(value Value) { values = append(values, value.byteArray()...) })
+		return encoding.FixedLenByteArrayValues(values, length)
+
+	default:
+		return encoding.Values{}
+	}
+}
+
+func (p *rowBufferPage) scan(f func(Value)) {
+	columnIndex := ^int16(p.column)
+
+	for _, row := range p.rows {
+		for _, value := range row {
+			if value.columnIndex == columnIndex {
+				f(value)
+			}
+		}
+	}
+}
+
+func (p *rowBufferPage) scanNonNull(f func(Value)) {
+	p.scan(func(value Value) {
+		if !value.isNull() {
+			f(value)
+		}
+	})
+}
+
+type rowBufferPageValueReader struct {
+	page        *rowBufferPage
+	rowIndex    int
+	valueIndex  int
+	columnIndex int16
+}
+
+func (r *rowBufferPageValueReader) ReadValues(values []Value) (n int, err error) {
+	for n < len(values) && r.rowIndex < len(r.page.rows) {
+		for n < len(values) && r.valueIndex < len(r.page.rows[r.rowIndex]) {
+			if v := r.page.rows[r.rowIndex][r.valueIndex]; v.columnIndex == r.columnIndex {
+				values[n] = v
+				n++
+			}
+			r.valueIndex++
+		}
+		r.rowIndex++
+		r.valueIndex = 0
+	}
+	if r.rowIndex == len(r.page.rows) {
+		err = io.EOF
+	}
+	return n, err
+}
+
+type rowBufferRows struct {
+	rows   []Row
+	index  int
+	schema *Schema
+}
+
+func (r *rowBufferRows) Close() error {
+	r.index = -1
+	return nil
+}
+
+func (r *rowBufferRows) Schema() *Schema {
+	return r.schema
+}
+
+func (r *rowBufferRows) SeekToRow(rowIndex int64) error {
+	if rowIndex < 0 {
+		return ErrSeekOutOfRange
+	}
+
+	if r.index < 0 {
+		return io.ErrClosedPipe
+	}
+
+	maxRowIndex := int64(len(r.rows))
+	if rowIndex > maxRowIndex {
+		rowIndex = maxRowIndex
+	}
+
+	r.index = int(rowIndex)
+	return nil
+}
+
+func (r *rowBufferRows) ReadRows(rows []Row) (n int, err error) {
+	if r.index < 0 {
+		return 0, io.EOF
+	}
+
+	if n = len(r.rows) - r.index; n > len(rows) {
+		n = len(rows)
+	}
+
+	for i, row := range r.rows[r.index : r.index+n] {
+		rows[i] = append(rows[i][:0], row...)
+	}
+
+	if r.index += n; r.index == len(r.rows) {
+		err = io.EOF
+	}
+
+	return n, err
+}
+
+var (
+	_ RowGroup       = (*RowBuffer[any])(nil)
+	_ RowWriter      = (*RowBuffer[any])(nil)
+	_ sort.Interface = (*RowBuffer[any])(nil)
+)
diff --git a/schema.go b/schema.go
index d79cbeb..888be71 100644
--- a/schema.go
+++ b/schema.go
@@ -302,6 +302,12 @@ func (s *Schema) Columns() [][]string {
 	return s.columns
 }
 
+// Comparator constructs a comparator function which orders rows according to
+// the list of sorting columns passed as arguments.
+func (s *Schema) Comparator(sortingColumns ...SortingColumn) func(Row, Row) int {
+	return compareRowsFuncOf(s, sortingColumns)
+}
+
 func (s *Schema) forEachNode(do func(name string, node Node)) {
 	forEachNodeOf(s.Name(), s, do)
 }
diff --git a/search.go b/search.go
index db0ec8c..14e604b 100644
--- a/search.go
+++ b/search.go
@@ -1,41 +1,5 @@
 package parquet
 
-// CompareNullsFirst constructs a comparison function which assumes that null
-// values are smaller than all other values.
-func CompareNullsFirst(cmp func(Value, Value) int) func(Value, Value) int {
-	return func(a, b Value) int {
-		switch {
-		case a.IsNull():
-			if b.IsNull() {
-				return 0
-			}
-			return -1
-		case b.IsNull():
-			return +1
-		default:
-			return cmp(a, b)
-		}
-	}
-}
-
-// CompareNullsLast constructs a comparison function which assumes that null
-// values are greater than all other values.
-func CompareNullsLast(cmp func(Value, Value) int) func(Value, Value) int {
-	return func(a, b Value) int {
-		switch {
-		case a.IsNull():
-			if b.IsNull() {
-				return 0
-			}
-			return +1
-		case b.IsNull():
-			return -1
-		default:
-			return cmp(a, b)
-		}
-	}
-}
-
 // Search is like Find, but uses the default ordering of the given type. Search
 // and Find are scoped to a given ColumnChunk and find the pages within a
 // ColumnChunk which might contain the result.  See Find for more details.
diff --git a/type.go b/type.go
index e997c98..82a06f9 100644
--- a/type.go
+++ b/type.go
@@ -257,7 +257,7 @@ func (t booleanType) String() string                           { return "BOOLEAN
 func (t booleanType) Kind() Kind                               { return Boolean }
 func (t booleanType) Length() int                              { return 1 }
 func (t booleanType) EstimateSize(n int) int64                 { return (int64(n) + 7) / 8 }
-func (t booleanType) Compare(a, b Value) int                   { return compareBool(a.Boolean(), b.Boolean()) }
+func (t booleanType) Compare(a, b Value) int                   { return compareBool(a.boolean(), b.boolean()) }
 func (t booleanType) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t booleanType) LogicalType() *format.LogicalType         { return nil }
 func (t booleanType) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -292,7 +292,7 @@ func (t booleanType) Decode(dst encoding.Values, src []byte, enc encoding.Encodi
 }
 
 func (t booleanType) AssignValue(dst reflect.Value, src Value) error {
-	v := src.Boolean()
+	v := src.boolean()
 	switch dst.Kind() {
 	case reflect.Bool:
 		dst.SetBool(v)
@@ -313,7 +313,7 @@ func (t int32Type) String() string                           { return "INT32" }
 func (t int32Type) Kind() Kind                               { return Int32 }
 func (t int32Type) Length() int                              { return 32 }
 func (t int32Type) EstimateSize(n int) int64                 { return 4 * int64(n) }
-func (t int32Type) Compare(a, b Value) int                   { return compareInt32(a.Int32(), b.Int32()) }
+func (t int32Type) Compare(a, b Value) int                   { return compareInt32(a.int32(), b.int32()) }
 func (t int32Type) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t int32Type) LogicalType() *format.LogicalType         { return nil }
 func (t int32Type) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -348,7 +348,7 @@ func (t int32Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 }
 
 func (t int32Type) AssignValue(dst reflect.Value, src Value) error {
-	v := src.Int32()
+	v := src.int32()
 	switch dst.Kind() {
 	case reflect.Int8, reflect.Int16, reflect.Int32:
 		dst.SetInt(int64(v))
@@ -371,7 +371,7 @@ func (t int64Type) String() string                           { return "INT64" }
 func (t int64Type) Kind() Kind                               { return Int64 }
 func (t int64Type) Length() int                              { return 64 }
 func (t int64Type) EstimateSize(n int) int64                 { return 8 * int64(n) }
-func (t int64Type) Compare(a, b Value) int                   { return compareInt64(a.Int64(), b.Int64()) }
+func (t int64Type) Compare(a, b Value) int                   { return compareInt64(a.int64(), b.int64()) }
 func (t int64Type) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t int64Type) LogicalType() *format.LogicalType         { return nil }
 func (t int64Type) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -406,7 +406,7 @@ func (t int64Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 }
 
 func (t int64Type) AssignValue(dst reflect.Value, src Value) error {
-	v := src.Int64()
+	v := src.int64()
 	switch dst.Kind() {
 	case reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Int:
 		dst.SetInt(v)
@@ -415,7 +415,6 @@ func (t int64Type) AssignValue(dst reflect.Value, src Value) error {
 	default:
 		dst.Set(reflect.ValueOf(v))
 	}
-
 	return nil
 }
 
@@ -430,7 +429,7 @@ func (t int96Type) String() string { return "INT96" }
 func (t int96Type) Kind() Kind                               { return Int96 }
 func (t int96Type) Length() int                              { return 96 }
 func (t int96Type) EstimateSize(n int) int64                 { return 12 * int64(n) }
-func (t int96Type) Compare(a, b Value) int                   { return compareInt96(a.Int96(), b.Int96()) }
+func (t int96Type) Compare(a, b Value) int                   { return compareInt96(a.int96(), b.int96()) }
 func (t int96Type) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t int96Type) LogicalType() *format.LogicalType         { return nil }
 func (t int96Type) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -480,7 +479,7 @@ func (t floatType) String() string                           { return "FLOAT" }
 func (t floatType) Kind() Kind                               { return Float }
 func (t floatType) Length() int                              { return 32 }
 func (t floatType) EstimateSize(n int) int64                 { return 4 * int64(n) }
-func (t floatType) Compare(a, b Value) int                   { return compareFloat32(a.Float(), b.Float()) }
+func (t floatType) Compare(a, b Value) int                   { return compareFloat32(a.float(), b.float()) }
 func (t floatType) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t floatType) LogicalType() *format.LogicalType         { return nil }
 func (t floatType) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -515,14 +514,13 @@ func (t floatType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 }
 
 func (t floatType) AssignValue(dst reflect.Value, src Value) error {
-	v := src.Float()
+	v := src.float()
 	switch dst.Kind() {
 	case reflect.Float32, reflect.Float64:
 		dst.SetFloat(float64(v))
 	default:
 		dst.Set(reflect.ValueOf(v))
 	}
-
 	return nil
 }
 
@@ -536,7 +534,7 @@ func (t doubleType) String() string                           { return "DOUBLE"
 func (t doubleType) Kind() Kind                               { return Double }
 func (t doubleType) Length() int                              { return 64 }
 func (t doubleType) EstimateSize(n int) int64                 { return 8 * int64(n) }
-func (t doubleType) Compare(a, b Value) int                   { return compareFloat64(a.Double(), b.Double()) }
+func (t doubleType) Compare(a, b Value) int                   { return compareFloat64(a.double(), b.double()) }
 func (t doubleType) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t doubleType) LogicalType() *format.LogicalType         { return nil }
 func (t doubleType) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -571,7 +569,7 @@ func (t doubleType) Decode(dst encoding.Values, src []byte, enc encoding.Encodin
 }
 
 func (t doubleType) AssignValue(dst reflect.Value, src Value) error {
-	v := src.Double()
+	v := src.double()
 	switch dst.Kind() {
 	case reflect.Float32, reflect.Float64:
 		dst.SetFloat(v)
@@ -592,7 +590,7 @@ func (t byteArrayType) String() string                           { return "BYTE_
 func (t byteArrayType) Kind() Kind                               { return ByteArray }
 func (t byteArrayType) Length() int                              { return 0 }
 func (t byteArrayType) EstimateSize(n int) int64                 { return 10 * int64(n) }
-func (t byteArrayType) Compare(a, b Value) int                   { return bytes.Compare(a.ByteArray(), b.ByteArray()) }
+func (t byteArrayType) Compare(a, b Value) int                   { return bytes.Compare(a.byteArray(), b.byteArray()) }
 func (t byteArrayType) ColumnOrder() *format.ColumnOrder         { return &typeDefinedColumnOrder }
 func (t byteArrayType) LogicalType() *format.LogicalType         { return nil }
 func (t byteArrayType) ConvertedType() *deprecated.ConvertedType { return nil }
@@ -627,7 +625,7 @@ func (t byteArrayType) Decode(dst encoding.Values, src []byte, enc encoding.Enco
 }
 
 func (t byteArrayType) AssignValue(dst reflect.Value, src Value) error {
-	v := src.ByteArray()
+	v := src.byteArray()
 	switch dst.Kind() {
 	case reflect.String:
 		dst.SetString(string(v))
@@ -658,7 +656,7 @@ func (t fixedLenByteArrayType) Length() int { return t.length }
 func (t fixedLenByteArrayType) EstimateSize(n int) int64 { return int64(t.length) * int64(n) }
 
 func (t fixedLenByteArrayType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
+	return bytes.Compare(a.byteArray(), b.byteArray())
 }
 
 func (t fixedLenByteArrayType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
@@ -698,7 +696,7 @@ func (t fixedLenByteArrayType) Decode(dst encoding.Values, src []byte, enc encod
 }
 
 func (t fixedLenByteArrayType) AssignValue(dst reflect.Value, src Value) error {
-	v := src.ByteArray()
+	v := src.byteArray()
 	switch dst.Kind() {
 	case reflect.Array:
 		if dst.Type().Elem().Kind() == reflect.Uint8 && dst.Len() == len(v) {
@@ -747,9 +745,7 @@ func (t be128Type) Length() int { return 16 }
 
 func (t be128Type) EstimateSize(n int) int64 { return 16 * int64(n) }
 
-func (t be128Type) Compare(a, b Value) int {
-	return compareBE128((*[16]byte)(a.ByteArray()), (*[16]byte)(b.ByteArray()))
-}
+func (t be128Type) Compare(a, b Value) int { return compareBE128(a.be128(), b.be128()) }
 
 func (t be128Type) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
 
@@ -869,16 +865,16 @@ func (t *intType) EstimateSize(n int) int64 { return int64(t.BitWidth/8) * int64
 
 func (t *intType) Compare(a, b Value) int {
 	if t.BitWidth == 64 {
-		i1 := a.Int64()
-		i2 := b.Int64()
+		i1 := a.int64()
+		i2 := b.int64()
 		if t.IsSigned {
 			return compareInt64(i1, i2)
 		} else {
 			return compareUint64(uint64(i1), uint64(i2))
 		}
 	} else {
-		i1 := a.Int32()
-		i2 := b.Int32()
+		i1 := a.int32()
+		i2 := b.int32()
 		if t.IsSigned {
 			return compareInt32(i1, i2)
 		} else {
@@ -1067,7 +1063,7 @@ func (t *stringType) Length() int { return 0 }
 func (t *stringType) EstimateSize(n int) int64 { return 10 * int64(n) }
 
 func (t *stringType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
+	return bytes.Compare(a.byteArray(), b.byteArray())
 }
 
 func (t *stringType) ColumnOrder() *format.ColumnOrder {
@@ -1137,9 +1133,7 @@ func (t *uuidType) Length() int { return 16 }
 
 func (t *uuidType) EstimateSize(n int) int64 { return 16 * int64(n) }
 
-func (t *uuidType) Compare(a, b Value) int {
-	return compareBE128((*[16]byte)(a.ByteArray()), (*[16]byte)(b.ByteArray()))
-}
+func (t *uuidType) Compare(a, b Value) int { return compareBE128(a.be128(), b.be128()) }
 
 func (t *uuidType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
 
@@ -1203,7 +1197,7 @@ func (t *enumType) Length() int { return 0 }
 func (t *enumType) EstimateSize(n int) int64 { return 10 * int64(n) }
 
 func (t *enumType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
+	return bytes.Compare(a.byteArray(), b.byteArray())
 }
 
 func (t *enumType) ColumnOrder() *format.ColumnOrder {
@@ -1274,7 +1268,7 @@ func (t *jsonType) Length() int { return 0 }
 func (t *jsonType) EstimateSize(n int) int64 { return 10 * int64(n) }
 
 func (t *jsonType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
+	return bytes.Compare(a.byteArray(), b.byteArray())
 }
 
 func (t *jsonType) ColumnOrder() *format.ColumnOrder {
@@ -1345,7 +1339,7 @@ func (t *bsonType) Length() int { return 0 }
 func (t *bsonType) EstimateSize(n int) int64 { return 10 * int64(n) }
 
 func (t *bsonType) Compare(a, b Value) int {
-	return bytes.Compare(a.ByteArray(), b.ByteArray())
+	return bytes.Compare(a.byteArray(), b.byteArray())
 }
 
 func (t *bsonType) ColumnOrder() *format.ColumnOrder {
@@ -1415,7 +1409,7 @@ func (t *dateType) Length() int { return 32 }
 
 func (t *dateType) EstimateSize(n int) int64 { return 4 * int64(n) }
 
-func (t *dateType) Compare(a, b Value) int { return compareInt32(a.Int32(), b.Int32()) }
+func (t *dateType) Compare(a, b Value) int { return compareInt32(a.int32(), b.int32()) }
 
 func (t *dateType) ColumnOrder() *format.ColumnOrder {
 	return &typeDefinedColumnOrder
@@ -1550,9 +1544,9 @@ func (t *timeType) EstimateSize(n int) int64 {
 
 func (t *timeType) Compare(a, b Value) int {
 	if t.useInt32() {
-		return compareInt32(a.Int32(), b.Int32())
+		return compareInt32(a.int32(), b.int32())
 	} else {
-		return compareInt64(a.Int64(), b.Int64())
+		return compareInt64(a.int64(), b.int64())
 	}
 }
 
@@ -1672,7 +1666,7 @@ func (t *timestampType) Length() int { return 64 }
 
 func (t *timestampType) EstimateSize(n int) int64 { return 8 * int64(n) }
 
-func (t *timestampType) Compare(a, b Value) int { return compareInt64(a.Int64(), b.Int64()) }
+func (t *timestampType) Compare(a, b Value) int { return compareInt64(a.int64(), b.int64()) }
 
 func (t *timestampType) ColumnOrder() *format.ColumnOrder { return &typeDefinedColumnOrder }
 
@@ -1730,7 +1724,7 @@ func (t *timestampType) AssignValue(dst reflect.Value, src Value) error {
 			unit = lt.Timestamp.Unit
 		}
 
-		nanos := src.Int64()
+		nanos := src.int64()
 		switch {
 		case unit.Millis != nil:
 			nanos = nanos * 1e6
@@ -1759,7 +1753,7 @@ func (t *timestampType) ConvertValue(val Value, typ Type) (Value, error) {
 
 	source := timeUnitDuration(sourceTs.Unit)
 	target := timeUnitDuration(t.Unit)
-	converted := val.Int64() * source.Nanoseconds() / target.Nanoseconds()
+	converted := val.int64() * source.Nanoseconds() / target.Nanoseconds()
 
 	return ValueOf(converted), nil
 }
diff --git a/value.go b/value.go
index 4a1ae6a..8b3a2b2 100644
--- a/value.go
+++ b/value.go
@@ -420,52 +420,69 @@ func makeValueByteArray(kind Kind, data *byte, size int) Value {
 	}
 }
 
+// These methods are internal versions of methods exported by the Value type,
+// they are usually inlined by the compiler and intended to be used inside the
+// parquet-go package because they tend to generate better code than their
+// exported counter part, which requires making a copy of the receiver.
+func (v *Value) isNull() bool            { return v.kind == 0 }
+func (v *Value) byte() byte              { return byte(v.u64) }
+func (v *Value) boolean() bool           { return v.u64 != 0 }
+func (v *Value) int32() int32            { return int32(v.u64) }
+func (v *Value) int64() int64            { return int64(v.u64) }
+func (v *Value) int96() deprecated.Int96 { return makeInt96(v.byteArray()) }
+func (v *Value) float() float32          { return math.Float32frombits(uint32(v.u64)) }
+func (v *Value) double() float64         { return math.Float64frombits(uint64(v.u64)) }
+func (v *Value) uint32() uint32          { return uint32(v.u64) }
+func (v *Value) uint64() uint64          { return v.u64 }
+func (v *Value) byteArray() []byte       { return unsafecast.Bytes(v.ptr, int(v.u64)) }
+func (v *Value) be128() *[16]byte        { return (*[16]byte)(unsafe.Pointer(v.ptr)) }
+func (v *Value) column() int             { return int(^v.columnIndex) }
+
 // Kind returns the kind of v, which represents its parquet physical type.
 func (v Value) Kind() Kind { return ^Kind(v.kind) }
 
 // IsNull returns true if v is the null value.
-func (v Value) IsNull() bool { return v.kind == 0 }
+func (v Value) IsNull() bool { return v.isNull() }
 
 // Byte returns v as a byte, which may truncate the underlying byte.
-func (v Value) Byte() byte { return byte(v.u64) }
+func (v Value) Byte() byte { return v.byte() }
 
 // Boolean returns v as a bool, assuming the underlying type is BOOLEAN.
-func (v Value) Boolean() bool { return v.u64 != 0 }
+func (v Value) Boolean() bool { return v.boolean() }
 
 // Int32 returns v as a int32, assuming the underlying type is INT32.
-func (v Value) Int32() int32 { return int32(v.u64) }
+func (v Value) Int32() int32 { return v.int32() }
 
 // Int64 returns v as a int64, assuming the underlying type is INT64.
-func (v Value) Int64() int64 { return int64(v.u64) }
+func (v Value) Int64() int64 { return v.int64() }
 
 // Int96 returns v as a int96, assuming the underlying type is INT96.
 func (v Value) Int96() deprecated.Int96 {
 	var val deprecated.Int96
-	if !v.IsNull() {
-		val = makeInt96(v.ByteArray())
+	if !v.isNull() {
+		val = v.int96()
 	}
-
 	return val
 }
 
 // Float returns v as a float32, assuming the underlying type is FLOAT.
-func (v Value) Float() float32 { return math.Float32frombits(uint32(v.u64)) }
+func (v Value) Float() float32 { return v.float() }
 
 // Double returns v as a float64, assuming the underlying type is DOUBLE.
-func (v Value) Double() float64 { return math.Float64frombits(v.u64) }
+func (v Value) Double() float64 { return v.double() }
 
 // Uint32 returns v as a uint32, assuming the underlying type is INT32.
-func (v Value) Uint32() uint32 { return uint32(v.u64) }
+func (v Value) Uint32() uint32 { return v.uint32() }
 
 // Uint64 returns v as a uint64, assuming the underlying type is INT64.
-func (v Value) Uint64() uint64 { return v.u64 }
+func (v Value) Uint64() uint64 { return v.uint64() }
 
 // ByteArray returns v as a []byte, assuming the underlying type is either
 // BYTE_ARRAY or FIXED_LEN_BYTE_ARRAY.
 //
 // The application must treat the returned byte slice as a read-only value,
 // mutating the content will result in undefined behaviors.
-func (v Value) ByteArray() []byte { return unsafe.Slice(v.ptr, int(v.u64)) }
+func (v Value) ByteArray() []byte { return v.byteArray() }
 
 // RepetitionLevel returns the repetition level of v.
 func (v Value) RepetitionLevel() int { return int(v.repetitionLevel) }
@@ -476,7 +493,7 @@ func (v Value) DefinitionLevel() int { return int(v.definitionLevel) }
 // Column returns the column index within the row that v was created from.
 //
 // Returns -1 if the value does not carry a column index.
-func (v Value) Column() int { return int(^v.columnIndex) }
+func (v Value) Column() int { return v.column() }
 
 // Bytes returns the binary representation of v.
 //
@@ -490,16 +507,16 @@ func (v Value) AppendBytes(b []byte) []byte {
 	buf := [8]byte{}
 	switch v.Kind() {
 	case Boolean:
-		binary.LittleEndian.PutUint32(buf[:4], uint32(v.u64))
+		binary.LittleEndian.PutUint32(buf[:4], v.uint32())
 		return append(b, buf[0])
 	case Int32, Float:
-		binary.LittleEndian.PutUint32(buf[:4], uint32(v.u64))
+		binary.LittleEndian.PutUint32(buf[:4], v.uint32())
 		return append(b, buf[:4]...)
 	case Int64, Double:
-		binary.LittleEndian.PutUint64(buf[:8], v.u64)
+		binary.LittleEndian.PutUint64(buf[:8], v.uint64())
 		return append(b, buf[:8]...)
 	case ByteArray, FixedLenByteArray, Int96:
-		return append(b, v.ByteArray()...)
+		return append(b, v.byteArray()...)
 	default:
 		return b
 	}
@@ -531,19 +548,19 @@ func (v Value) Format(w fmt.State, r rune) {
 		if w.Flag('+') {
 			io.WriteString(w, "C:")
 		}
-		fmt.Fprint(w, v.Column())
+		fmt.Fprint(w, v.column())
 
 	case 'd':
 		if w.Flag('+') {
 			io.WriteString(w, "D:")
 		}
-		fmt.Fprint(w, v.DefinitionLevel())
+		fmt.Fprint(w, v.definitionLevel)
 
 	case 'r':
 		if w.Flag('+') {
 			io.WriteString(w, "R:")
 		}
-		fmt.Fprint(w, v.RepetitionLevel())
+		fmt.Fprint(w, v.repetitionLevel)
 
 	case 'q':
 		if w.Flag('+') {
@@ -551,7 +568,7 @@ func (v Value) Format(w fmt.State, r rune) {
 		}
 		switch v.Kind() {
 		case ByteArray, FixedLenByteArray:
-			fmt.Fprintf(w, "%q", v.ByteArray())
+			fmt.Fprintf(w, "%q", v.byteArray())
 		default:
 			fmt.Fprintf(w, `"%s"`, v)
 		}
@@ -562,19 +579,19 @@ func (v Value) Format(w fmt.State, r rune) {
 		}
 		switch v.Kind() {
 		case Boolean:
-			fmt.Fprint(w, v.Boolean())
+			fmt.Fprint(w, v.boolean())
 		case Int32:
-			fmt.Fprint(w, v.Int32())
+			fmt.Fprint(w, v.int32())
 		case Int64:
-			fmt.Fprint(w, v.Int64())
+			fmt.Fprint(w, v.int64())
 		case Int96:
-			fmt.Fprint(w, v.Int96())
+			fmt.Fprint(w, v.int96())
 		case Float:
-			fmt.Fprint(w, v.Float())
+			fmt.Fprint(w, v.float())
 		case Double:
-			fmt.Fprint(w, v.Double())
+			fmt.Fprint(w, v.double())
 		case ByteArray, FixedLenByteArray:
-			w.Write(v.ByteArray())
+			w.Write(v.byteArray())
 		default:
 			io.WriteString(w, "<null>")
 		}
@@ -595,33 +612,31 @@ func (v Value) Format(w fmt.State, r rune) {
 func (v Value) String() string {
 	switch v.Kind() {
 	case Boolean:
-		return strconv.FormatBool(v.Boolean())
+		return strconv.FormatBool(v.boolean())
 	case Int32:
-		return strconv.FormatInt(int64(v.Int32()), 10)
+		return strconv.FormatInt(int64(v.int32()), 10)
 	case Int64:
-		return strconv.FormatInt(v.Int64(), 10)
+		return strconv.FormatInt(v.int64(), 10)
 	case Int96:
 		return v.Int96().String()
 	case Float:
-		return strconv.FormatFloat(float64(v.Float()), 'g', -1, 32)
+		return strconv.FormatFloat(float64(v.float()), 'g', -1, 32)
 	case Double:
-		return strconv.FormatFloat(v.Double(), 'g', -1, 32)
+		return strconv.FormatFloat(v.double(), 'g', -1, 32)
 	case ByteArray, FixedLenByteArray:
 		// As an optimizations for the common case of using String on UTF8
 		// columns we convert the byte array to a string without copying the
 		// underlying data to a new memory location. This is safe as long as the
 		// application respects the requirement to not mutate the byte slices
 		// returned when calling ByteArray.
-		return unsafecast.BytesToString(v.ByteArray())
+		return unsafecast.BytesToString(v.byteArray())
 	default:
 		return "<null>"
 	}
 }
 
 // GoString returns a Go value string representation of v.
-func (v Value) GoString() string {
-	return fmt.Sprintf("%#v", v)
-}
+func (v Value) GoString() string { return fmt.Sprintf("%#v", v) }
 
 // Level returns v with the repetition level, definition level, and column index
 // set to the values passed as arguments.
@@ -638,8 +653,7 @@ func (v Value) Level(repetitionLevel, definitionLevel, columnIndex int) Value {
 func (v Value) Clone() Value {
 	switch k := v.Kind(); k {
 	case ByteArray, FixedLenByteArray:
-		b := copyBytes(v.ByteArray())
-		v.ptr = unsafecast.AddressOfBytes(b)
+		v.ptr = unsafecast.AddressOfBytes(copyBytes(v.byteArray()))
 	}
 	return v
 }
@@ -681,7 +695,7 @@ func parseValue(kind Kind, data []byte) (val Value, err error) {
 	case ByteArray, FixedLenByteArray:
 		val = makeValueBytes(kind, data)
 	}
-	if val.IsNull() {
+	if val.isNull() {
 		err = fmt.Errorf("cannot decode %s value from input of length %d", kind, len(data))
 	}
 	return val, err
@@ -705,21 +719,21 @@ func Equal(v1, v2 Value) bool {
 	if v1.kind != v2.kind {
 		return false
 	}
-	switch v1.Kind() {
+	switch ^Kind(v1.kind) {
 	case Boolean:
-		return v1.Boolean() == v2.Boolean()
+		return v1.boolean() == v2.boolean()
 	case Int32:
-		return v1.Int32() == v2.Int32()
+		return v1.int32() == v2.int32()
 	case Int64:
-		return v1.Int64() == v2.Int64()
+		return v1.int64() == v2.int64()
 	case Int96:
-		return v1.Int96() == v2.Int96()
+		return v1.int96() == v2.int96()
 	case Float:
-		return v1.Float() == v2.Float()
+		return v1.float() == v2.float()
 	case Double:
-		return v1.Double() == v2.Double()
+		return v1.double() == v2.double()
 	case ByteArray, FixedLenByteArray:
-		return bytes.Equal(v1.ByteArray(), v2.ByteArray())
+		return bytes.Equal(v1.byteArray(), v2.byteArray())
 	case -1: // null
 		return true
 	default:


Test patch:
diff --git a/buffer_go18_test.go b/buffer_go18_test.go
index 5833d6c..18acf64 100644
--- a/buffer_go18_test.go
+++ b/buffer_go18_test.go
@@ -3,11 +3,13 @@
 package parquet_test
 
 import (
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
 	"math/rand"
 	"reflect"
+	"sort"
 	"testing"
 
 	"github.com/segmentio/parquet-go"
@@ -219,3 +221,44 @@ func TestIssue347(t *testing.T) {
 	}()
 	_ = parquet.NewGenericBuffer[any]()
 }
+
+func BenchmarkSortGenericBuffer(b *testing.B) {
+	type Row struct {
+		I0 int64
+		I1 int64
+		I2 int64
+		I3 int64
+		I4 int64
+		I5 int64
+		I6 int64
+		I7 int64
+		I8 int64
+		I9 int64
+		ID [16]byte
+	}
+
+	buf := parquet.NewGenericBuffer[Row](
+		parquet.SortingColumns(
+			parquet.Ascending("ID"),
+		),
+	)
+
+	rows := make([]Row, 10e3)
+	prng := rand.New(rand.NewSource(0))
+
+	for i := range rows {
+		binary.LittleEndian.PutUint64(rows[i].ID[:8], uint64(i))
+		binary.LittleEndian.PutUint64(rows[i].ID[8:], ^uint64(i))
+	}
+
+	buf.Write(rows)
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 10; j++ {
+			buf.Swap(prng.Intn(len(rows)), prng.Intn(len(rows)))
+		}
+
+		sort.Sort(buf)
+	}
+}
diff --git a/compare_test.go b/compare_test.go
index 00e7708..73021ba 100644
--- a/compare_test.go
+++ b/compare_test.go
@@ -2,6 +2,28 @@ package parquet
 
 import "testing"
 
+func assertCompare(t *testing.T, a, b Value, cmp func(Value, Value) int, want int) {
+	if got := cmp(a, b); got != want {
+		t.Errorf("compare(%v, %v): got=%d want=%d", a, b, got, want)
+	}
+}
+
+func TestCompareNullsFirst(t *testing.T) {
+	cmp := CompareNullsFirst(Int32Type.Compare)
+	assertCompare(t, Value{}, Value{}, cmp, 0)
+	assertCompare(t, Value{}, ValueOf(int32(0)), cmp, -1)
+	assertCompare(t, ValueOf(int32(0)), Value{}, cmp, +1)
+	assertCompare(t, ValueOf(int32(0)), ValueOf(int32(1)), cmp, -1)
+}
+
+func TestCompareNullsLast(t *testing.T) {
+	cmp := CompareNullsLast(Int32Type.Compare)
+	assertCompare(t, Value{}, Value{}, cmp, 0)
+	assertCompare(t, Value{}, ValueOf(int32(0)), cmp, +1)
+	assertCompare(t, ValueOf(int32(0)), Value{}, cmp, -1)
+	assertCompare(t, ValueOf(int32(0)), ValueOf(int32(1)), cmp, -1)
+}
+
 func BenchmarkCompareBE128(b *testing.B) {
 	v1 := [16]byte{}
 	v2 := [16]byte{}
diff --git a/row_buffer_test.go b/row_buffer_test.go
new file mode 100644
index 0000000..82c3427
--- /dev/null
+++ b/row_buffer_test.go
@@ -0,0 +1,296 @@
+//go:build go1.18
+
+package parquet_test
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"math/rand"
+	"reflect"
+	"sort"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+	"github.com/segmentio/parquet-go/encoding"
+)
+
+func TestRowBuffer(t *testing.T) {
+	testRowBuffer[booleanColumn](t)
+	testRowBuffer[int32Column](t)
+	testRowBuffer[int64Column](t)
+	testRowBuffer[int96Column](t)
+	testRowBuffer[floatColumn](t)
+	testRowBuffer[doubleColumn](t)
+	testRowBuffer[byteArrayColumn](t)
+	testRowBuffer[fixedLenByteArrayColumn](t)
+	testRowBuffer[stringColumn](t)
+	testRowBuffer[indexedStringColumn](t)
+	testRowBuffer[uuidColumn](t)
+	testRowBuffer[timeColumn](t)
+	testRowBuffer[timeInMillisColumn](t)
+	testRowBuffer[mapColumn](t)
+	testRowBuffer[decimalColumn](t)
+	testRowBuffer[addressBook](t)
+	testRowBuffer[contact](t)
+	testRowBuffer[listColumn2](t)
+	testRowBuffer[listColumn1](t)
+	testRowBuffer[listColumn0](t)
+	testRowBuffer[nestedListColumn1](t)
+	testRowBuffer[nestedListColumn](t)
+	testRowBuffer[*contact](t)
+	testRowBuffer[paddedBooleanColumn](t)
+	testRowBuffer[optionalInt32Column](t)
+	testRowBuffer[repeatedInt32Column](t)
+
+	for _, test := range bufferTests {
+		t.Run(test.scenario, func(t *testing.T) {
+			for _, mod := range [...]struct {
+				scenario string
+				function func(parquet.Node) parquet.Node
+			}{
+				{scenario: "optional", function: parquet.Optional},
+				{scenario: "repeated", function: parquet.Repeated},
+				{scenario: "required", function: parquet.Required},
+			} {
+				t.Run(mod.scenario, func(t *testing.T) {
+					for _, ordering := range [...]struct {
+						scenario string
+						sorting  parquet.SortingColumn
+						sortFunc func(parquet.Type, []parquet.Value)
+					}{
+						{scenario: "unordered", sorting: nil, sortFunc: unordered},
+						{scenario: "ascending", sorting: parquet.Ascending("data"), sortFunc: ascending},
+						{scenario: "descending", sorting: parquet.Descending("data"), sortFunc: descending},
+					} {
+						t.Run(ordering.scenario, func(t *testing.T) {
+							schema := parquet.NewSchema("test", parquet.Group{
+								"data": mod.function(parquet.Leaf(test.typ)),
+							})
+
+							options := []parquet.RowGroupOption{
+								schema,
+							}
+
+							if ordering.sorting != nil {
+								options = append(options, parquet.SortingColumns(ordering.sorting))
+							}
+
+							content := new(bytes.Buffer)
+							buffer := parquet.NewRowBuffer[any](options...)
+
+							for _, values := range test.values {
+								t.Run("", func(t *testing.T) {
+									defer content.Reset()
+									defer buffer.Reset()
+									fields := schema.Fields()
+									testRowBufferAny(t, fields[0], buffer, &parquet.Plain, values, ordering.sortFunc)
+								})
+							}
+						})
+					}
+				})
+			}
+		})
+	}
+}
+
+func testRowBuffer[Row any](t *testing.T) {
+	var model Row
+	t.Run(reflect.TypeOf(model).Name(), func(t *testing.T) {
+		err := quickCheck(func(rows []Row) bool {
+			if len(rows) == 0 {
+				return true // TODO: fix support for parquet files with zero rows
+			}
+			if err := testRowBufferRows(rows); err != nil {
+				t.Error(err)
+				return false
+			}
+			return true
+		})
+		if err != nil {
+			t.Error(err)
+		}
+	})
+}
+
+func testRowBufferRows[Row any](rows []Row) error {
+	setNullPointers(rows)
+	buffer := parquet.NewRowBuffer[Row]()
+	_, err := buffer.Write(rows)
+	if err != nil {
+		return err
+	}
+	reader := parquet.NewGenericRowGroupReader[Row](buffer)
+	result := make([]Row, len(rows))
+	n, err := reader.Read(result)
+	if err != nil && !errors.Is(err, io.EOF) {
+		return err
+	}
+	if n < len(rows) {
+		return fmt.Errorf("not enough values were read: want=%d got=%d", len(rows), n)
+	}
+	if !reflect.DeepEqual(rows, result) {
+		return fmt.Errorf("rows mismatch:\nwant: %#v\ngot:  %#v", rows, result)
+	}
+	return nil
+}
+
+func testRowBufferAny(t *testing.T, node parquet.Node, buffer *parquet.RowBuffer[any], encoding encoding.Encoding, values []any, sortFunc sortFunc) {
+	repetitionLevel := 0
+	definitionLevel := 0
+	if !node.Required() {
+		definitionLevel = 1
+	}
+
+	minValue := parquet.Value{}
+	maxValue := parquet.Value{}
+	batch := make([]parquet.Value, len(values))
+	for i := range values {
+		batch[i] = parquet.ValueOf(values[i]).Level(repetitionLevel, definitionLevel, 0)
+	}
+
+	for i := range batch {
+		_, err := buffer.WriteRows([]parquet.Row{batch[i : i+1]})
+		if err != nil {
+			t.Fatalf("writing value to row group: %v", err)
+		}
+	}
+
+	numRows := buffer.NumRows()
+	if numRows != int64(len(batch)) {
+		t.Fatalf("number of rows mismatch: want=%d got=%d", len(batch), numRows)
+	}
+
+	typ := node.Type()
+	for _, value := range batch {
+		if minValue.IsNull() || typ.Compare(value, minValue) < 0 {
+			minValue = value
+		}
+		if maxValue.IsNull() || typ.Compare(value, maxValue) > 0 {
+			maxValue = value
+		}
+	}
+
+	sortFunc(typ, batch)
+	sort.Sort(buffer)
+
+	pages := buffer.ColumnChunks()[0].Pages()
+	page, err := pages.ReadPage()
+	defer pages.Close()
+
+	if err == io.EOF {
+		if numRows != 0 {
+			t.Fatalf("no pages found in row buffer despite having %d rows", numRows)
+		} else {
+			return
+		}
+	}
+
+	numValues := page.NumValues()
+	if numValues != int64(len(batch)) {
+		t.Fatalf("number of values mistmatch: want=%d got=%d", len(batch), numValues)
+	}
+
+	numNulls := page.NumNulls()
+	if numNulls != 0 {
+		t.Fatalf("number of nulls mismatch: want=0 got=%d", numNulls)
+	}
+
+	min, max, hasBounds := page.Bounds()
+	if !hasBounds && numRows > 0 {
+		t.Fatal("page bounds are missing")
+	}
+	if !parquet.Equal(min, minValue) {
+		t.Fatalf("min value mismatch: want=%v got=%v", minValue, min)
+	}
+	if !parquet.Equal(max, maxValue) {
+		t.Fatalf("max value mismatch: want=%v got=%v", maxValue, max)
+	}
+
+	// We write a single value per row, so num values = num rows for all pages
+	// including repeated ones, which makes it OK to slice the pages using the
+	// number of values as a proxy for the row indexes.
+	halfValues := numValues / 2
+
+	for _, test := range [...]struct {
+		scenario string
+		values   []parquet.Value
+		reader   parquet.ValueReader
+	}{
+		{"page", batch, page.Values()},
+		{"head", batch[:halfValues], page.Slice(0, halfValues).Values()},
+		{"tail", batch[halfValues:], page.Slice(halfValues, numValues).Values()},
+	} {
+		v := [1]parquet.Value{}
+		i := 0
+
+		for {
+			n, err := test.reader.ReadValues(v[:])
+			if n > 0 {
+				if n != 1 {
+					t.Fatalf("reading value from %q reader returned the wrong count: want=1 got=%d", test.scenario, n)
+				}
+				if i < len(test.values) {
+					if !parquet.Equal(v[0], test.values[i]) {
+						t.Fatalf("%q value at index %d mismatches: want=%v got=%v", test.scenario, i, test.values[i], v[0])
+					}
+				}
+				i++
+			}
+			if err != nil {
+				if err == io.EOF {
+					break
+				}
+				t.Fatalf("reading value from %q reader: %v", test.scenario, err)
+			}
+		}
+
+		if i != len(test.values) {
+			t.Errorf("wrong number of values read from %q reader: want=%d got=%d", test.scenario, len(test.values), i)
+		}
+	}
+}
+
+func BenchmarkSortRowBuffer(b *testing.B) {
+	type Row struct {
+		I0 int64
+		I1 int64
+		I2 int64
+		I3 int64
+		I4 int64
+		I5 int64
+		I6 int64
+		I7 int64
+		I8 int64
+		I9 int64
+		ID [16]byte
+	}
+
+	buf := parquet.NewRowBuffer[Row](
+		parquet.SortingColumns(
+			parquet.Ascending("ID"),
+		),
+	)
+
+	rows := make([]Row, 10e3)
+	prng := rand.New(rand.NewSource(0))
+
+	for i := range rows {
+		binary.LittleEndian.PutUint64(rows[i].ID[:8], uint64(i))
+		binary.LittleEndian.PutUint64(rows[i].ID[8:], ^uint64(i))
+	}
+
+	buf.Write(rows)
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 10; j++ {
+			buf.Swap(prng.Intn(len(rows)), prng.Intn(len(rows)))
+		}
+
+		sort.Sort(buf)
+	}
+}
diff --git a/search_test.go b/search_test.go
index e10fd99..e724416 100644
--- a/search_test.go
+++ b/search_test.go
@@ -6,28 +6,6 @@ import (
 	"github.com/segmentio/parquet-go"
 )
 
-func assertCompare(t *testing.T, a, b parquet.Value, cmp func(parquet.Value, parquet.Value) int, want int) {
-	if got := cmp(a, b); got != want {
-		t.Errorf("compare(%v, %v): got=%d want=%d", a, b, got, want)
-	}
-}
-
-func TestCompareNullsFirst(t *testing.T) {
-	cmp := parquet.CompareNullsFirst(parquet.Int32Type.Compare)
-	assertCompare(t, parquet.Value{}, parquet.Value{}, cmp, 0)
-	assertCompare(t, parquet.Value{}, parquet.ValueOf(int32(0)), cmp, -1)
-	assertCompare(t, parquet.ValueOf(int32(0)), parquet.Value{}, cmp, +1)
-	assertCompare(t, parquet.ValueOf(int32(0)), parquet.ValueOf(int32(1)), cmp, -1)
-}
-
-func TestCompareNullsLast(t *testing.T) {
-	cmp := parquet.CompareNullsLast(parquet.Int32Type.Compare)
-	assertCompare(t, parquet.Value{}, parquet.Value{}, cmp, 0)
-	assertCompare(t, parquet.Value{}, parquet.ValueOf(int32(0)), cmp, +1)
-	assertCompare(t, parquet.ValueOf(int32(0)), parquet.Value{}, cmp, -1)
-	assertCompare(t, parquet.ValueOf(int32(0)), parquet.ValueOf(int32(1)), cmp, -1)
-}
-
 func TestSearchBinary(t *testing.T) {
 	testSearch(t, [][]int32{
 		{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},