Repository: segmentio/parquet-go
Pull request #386 (2022-10-25 20:10:15)

set byte slice capacity when growing output buffers
I have a theory that #372 might be caused by growing memory allocations in small increments due to calling `make` with only the slice size and leaving the capacity allocation up to the runtime.

This PR modifies the `encoding` packages to always create slices with twice the capacity as the target size to ensure that we have room for growth when the memory areas are reused.

@mdisibio let me know if the change helps address the issue that you originally reported!


Patch:
diff --git a/column.go b/column.go
index d511623..a376c7d 100644
--- a/column.go
+++ b/column.go
@@ -633,7 +633,9 @@ func (c *Column) decodeDataPage(header DataPageHeader, numValues int, repetition
 	pageValues = vbuf.data
 
 	pageKind := pageType.Kind()
-	if pageKind == ByteArray {
+
+	// Page offsets not needed when dictionary-encoded
+	if pageKind == ByteArray && !isDictionaryEncoding(pageEncoding) {
 		obuf = buffers.get(4 * (numValues + 1))
 		defer obuf.unref()
 		pageOffsets = unsafecast.BytesToUint32(obuf.data)
diff --git a/config.go b/config.go
index 4e2efc2..b18a13a 100644
--- a/config.go
+++ b/config.go
@@ -2,6 +2,7 @@ package parquet
 
 import (
 	"fmt"
+	"math"
 	"runtime/debug"
 	"strings"
 	"sync"
@@ -18,6 +19,7 @@ const (
 	DefaultDataPageStatistics   = false
 	DefaultSkipPageIndex        = false
 	DefaultSkipBloomFilters     = false
+	DefaultMaxRowsPerRowGroup   = math.MaxInt64
 )
 
 const (
@@ -188,6 +190,7 @@ type WriterConfig struct {
 	WriteBufferSize      int
 	DataPageVersion      int
 	DataPageStatistics   bool
+	MaxRowsPerRowGroup   int64
 	KeyValueMetadata     map[string]string
 	Schema               *Schema
 	SortingColumns       []SortingColumn
@@ -206,6 +209,7 @@ func DefaultWriterConfig() *WriterConfig {
 		WriteBufferSize:      DefaultWriteBufferSize,
 		DataPageVersion:      DefaultDataPageVersion,
 		DataPageStatistics:   DefaultDataPageStatistics,
+		MaxRowsPerRowGroup:   DefaultMaxRowsPerRowGroup,
 	}
 }
 
@@ -246,6 +250,7 @@ func (c *WriterConfig) ConfigureWriter(config *WriterConfig) {
 		WriteBufferSize:      coalesceInt(c.WriteBufferSize, config.WriteBufferSize),
 		DataPageVersion:      coalesceInt(c.DataPageVersion, config.DataPageVersion),
 		DataPageStatistics:   config.DataPageStatistics,
+		MaxRowsPerRowGroup:   config.MaxRowsPerRowGroup,
 		KeyValueMetadata:     keyValueMetadata,
 		Schema:               coalesceSchema(c.Schema, config.Schema),
 		SortingColumns:       coalesceSortingColumns(c.SortingColumns, config.SortingColumns),
@@ -370,8 +375,8 @@ func SkipBloomFilters(skip bool) FileOption {
 // like network storage it can be advantageous to increase this value.
 //
 // Defaults to 4096.
-func ReadBufferSize(sz int) FileOption {
-	return fileOption(func(config *FileConfig) { config.ReadBufferSize = sz })
+func ReadBufferSize(size int) FileOption {
+	return fileOption(func(config *FileConfig) { config.ReadBufferSize = size })
 }
 
 // PageBufferSize configures the size of column page buffers on parquet writers.
@@ -397,6 +402,19 @@ func WriteBufferSize(size int) WriterOption {
 	return writerOption(func(config *WriterConfig) { config.WriteBufferSize = size })
 }
 
+// MaxRowsPerRowGroup configures the maximum number of rows that a writer will
+// produce in each row group.
+//
+// This limit is useful to control size of row groups in both number of rows and
+// byte size. While controlling the byte size of a row group is difficult to
+// achieve with parquet due to column encoding and compression, the number of
+// rows remains a useful proxy.
+//
+// Defaults to unlimited.
+func MaxRowsPerRowGroup(numRows int64) WriterOption {
+	return writerOption(func(config *WriterConfig) { config.MaxRowsPerRowGroup = numRows })
+}
+
 // CreatedBy creates a configuration option which sets the name of the
 // application that created a parquet file.
 //
diff --git a/encoding/bitpacked/bitpacked.go b/encoding/bitpacked/bitpacked.go
index f49cdb8..4f573ea 100644
--- a/encoding/bitpacked/bitpacked.go
+++ b/encoding/bitpacked/bitpacked.go
@@ -44,7 +44,7 @@ func encodeLevels(dst, src []byte, bitWidth uint) ([]byte, error) {
 	c := n + 1
 
 	if cap(dst) < c {
-		dst = make([]byte, c)
+		dst = make([]byte, c, 2*c)
 	} else {
 		dst = dst[:c]
 		for i := range dst {
@@ -80,7 +80,7 @@ func decodeLevels(dst, src []byte, bitWidth uint) ([]byte, error) {
 	}
 
 	if cap(dst) < numValues {
-		dst = make([]byte, numValues)
+		dst = make([]byte, numValues, 2*numValues)
 	} else {
 		dst = dst[:numValues]
 		for i := range dst {
diff --git a/encoding/delta/length_byte_array.go b/encoding/delta/length_byte_array.go
index ad83c5d..5cac384 100644
--- a/encoding/delta/length_byte_array.go
+++ b/encoding/delta/length_byte_array.go
@@ -46,7 +46,7 @@ func (e *LengthByteArrayEncoding) DecodeByteArray(dst []byte, src []byte, offset
 	}
 
 	if size := len(length.values) + 1; cap(offsets) < size {
-		offsets = make([]uint32, size)
+		offsets = make([]uint32, size, 2*size)
 	} else {
 		offsets = offsets[:size]
 	}
diff --git a/errors.go b/errors.go
index f3dd009..08af7e1 100644
--- a/errors.go
+++ b/errors.go
@@ -52,7 +52,7 @@ var (
 
 	// ErrTooManyRowGroups is returned when attempting to generate a parquet
 	// file with more than MaxRowGroups row groups.
-	ErrTooManyRowGroups = errors.New("the limit of 65535 row groups has been reached")
+	ErrTooManyRowGroups = errors.New("the limit of 32767 row groups has been reached")
 )
 
 type errno int
diff --git a/row.go b/row.go
index b716bb5..602fdb0 100644
--- a/row.go
+++ b/row.go
@@ -739,10 +739,13 @@ func reconstructFuncOfGroup(columnIndex int16, node Node) (int16, reconstructFun
 
 //go:noinline
 func reconstructFuncOfLeaf(columnIndex int16, node Node) (int16, reconstructFunc) {
+	typ := node.Type()
 	return columnIndex + 1, func(value reflect.Value, _ levels, row Row) (Row, error) {
 		if !row.startsWith(columnIndex) {
 			return row, fmt.Errorf("no values found in parquet row for column %d", columnIndex)
 		}
-		return row[1:], assignValue(value, row[0])
+
+		err := typ.AssignValue(value, row[0])
+		return row[1:], err
 	}
 }
diff --git a/search.go b/search.go
index bf23c52..db0ec8c 100644
--- a/search.go
+++ b/search.go
@@ -75,33 +75,65 @@ func Find(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
 
 func binarySearch(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
 	n := index.NumPages()
-	i := 0
-	j := n
+	curIdx := 0
+	topIdx := n
 
-	for (j - i) > 1 {
-		k := ((j - i) / 2) + i
-		c := cmp(value, index.MinValue(k))
+	// while there's at least one more page to check
+	for (topIdx - curIdx) > 1 {
+
+		// nextIdx is set to halfway between curIdx and topIdx
+		nextIdx := ((topIdx - curIdx) / 2) + curIdx
+
+		smallerThanMin := cmp(value, index.MinValue(nextIdx))
 
 		switch {
-		case c < 0:
-			j = k
-		case c > 0:
-			i = k
-		default:
-			return k
+		// search below pages[nextIdx]
+		case smallerThanMin < 0:
+			topIdx = nextIdx
+		// search pages[nextIdx] and above
+		case smallerThanMin > 0:
+			curIdx = nextIdx
+		case smallerThanMin == 0:
+			// this case is hit when winValue == value of nextIdx
+			// we must check below this index to find if there's
+			// another page before this.
+			// e.g. searching for first page 3 is in:
+			// [1,2,3]
+			// [3,4,5]
+			// [6,7,8]
+
+			// if the page proceeding this has a maxValue matching the value we're
+			// searching, continue the search.
+			// otherwise, we can return early
+			//
+			// cases covered by else block
+			// if cmp(value, index.MaxValue(nextIdx-1)) < 0: the value is only in this page
+			// if cmp(value, index.MaxValue(nextIdx-1)) > 0: we've got a sorting problem with overlapping pages
+			//
+			// bounds check not needed for nextIdx-1 because nextIdx is guaranteed to be at least curIdx + 1
+			// line 82 & 85 above
+			if cmp(value, index.MaxValue(nextIdx-1)) == 0 {
+				topIdx = nextIdx
+			} else {
+				return nextIdx
+			}
 		}
 	}
 
-	if i < n {
-		min := index.MinValue(i)
-		max := index.MaxValue(i)
+	// last page check, if it wasn't explicitly found above
+	if curIdx < n {
+
+		// check pages[curIdx] for value
+		min := index.MinValue(curIdx)
+		max := index.MaxValue(curIdx)
 
-		if cmp(value, min) < 0 || cmp(max, value) < 0 {
-			i = n
+		// if value is not in pages[curIdx], then it's not in this columnChunk
+		if cmp(value, min) < 0 || cmp(value, max) > 0 {
+			curIdx = n
 		}
 	}
 
-	return i
+	return curIdx
 }
 
 func linearSearch(index ColumnIndex, value Value, cmp func(Value, Value) int) int {
diff --git a/type.go b/type.go
index 312989f..556dde5 100644
--- a/type.go
+++ b/type.go
@@ -4,11 +4,14 @@ import (
 	"bytes"
 	"fmt"
 	"math/bits"
+	"reflect"
 	"time"
+	"unsafe"
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
 	"github.com/segmentio/parquet-go/format"
+	"github.com/segmentio/parquet-go/internal/unsafecast"
 )
 
 // Kind is an enumeration type representing the physical types supported by the
@@ -184,6 +187,10 @@ type Type interface {
 	// buffer passed as first argument by dispatching the call to one of the
 	// encoding methods.
 	Decode(dst encoding.Values, src []byte, enc encoding.Encoding) (encoding.Values, error)
+
+	// Assigns a Parquet value to a Go value. Returns an error if conversion is
+	// not possible.
+	AssignValue(dst reflect.Value, src Value) error
 }
 
 var (
@@ -278,6 +285,18 @@ func (t booleanType) Decode(dst encoding.Values, src []byte, enc encoding.Encodi
 	return encoding.DecodeBoolean(dst, src, enc)
 }
 
+func (t booleanType) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Boolean()
+	switch dst.Kind() {
+	case reflect.Bool:
+		dst.SetBool(v)
+	default:
+		dst.Set(reflect.ValueOf(v))
+	}
+
+	return nil
+}
+
 type int32Type struct{}
 
 func (t int32Type) String() string                           { return "INT32" }
@@ -318,6 +337,20 @@ func (t int32Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeInt32(dst, src, enc)
 }
 
+func (t int32Type) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Int32()
+	switch dst.Kind() {
+	case reflect.Int8, reflect.Int16, reflect.Int32:
+		dst.SetInt(int64(v))
+	case reflect.Uint8, reflect.Uint16, reflect.Uint32:
+		dst.SetUint(uint64(v))
+	default:
+		dst.Set(reflect.ValueOf(v))
+	}
+
+	return nil
+}
+
 type int64Type struct{}
 
 func (t int64Type) String() string                           { return "INT64" }
@@ -358,6 +391,20 @@ func (t int64Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeInt64(dst, src, enc)
 }
 
+func (t int64Type) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Int64()
+	switch dst.Kind() {
+	case reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Int:
+		dst.SetInt(v)
+	case reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uint, reflect.Uintptr:
+		dst.SetUint(uint64(v))
+	default:
+		dst.Set(reflect.ValueOf(v))
+	}
+
+	return nil
+}
+
 type int96Type struct{}
 
 func (t int96Type) String() string { return "INT96" }
@@ -399,6 +446,12 @@ func (t int96Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeInt96(dst, src, enc)
 }
 
+func (t int96Type) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Int96()
+	dst.Set(reflect.ValueOf(v))
+	return nil
+}
+
 type floatType struct{}
 
 func (t floatType) String() string                           { return "FLOAT" }
@@ -439,6 +492,18 @@ func (t floatType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeFloat(dst, src, enc)
 }
 
+func (t floatType) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Float()
+	switch dst.Kind() {
+	case reflect.Float32, reflect.Float64:
+		dst.SetFloat(float64(v))
+	default:
+		dst.Set(reflect.ValueOf(v))
+	}
+
+	return nil
+}
+
 type doubleType struct{}
 
 func (t doubleType) String() string                           { return "DOUBLE" }
@@ -479,6 +544,18 @@ func (t doubleType) Decode(dst encoding.Values, src []byte, enc encoding.Encodin
 	return encoding.DecodeDouble(dst, src, enc)
 }
 
+func (t doubleType) AssignValue(dst reflect.Value, src Value) error {
+	v := src.Double()
+	switch dst.Kind() {
+	case reflect.Float32, reflect.Float64:
+		dst.SetFloat(v)
+	default:
+		dst.Set(reflect.ValueOf(v))
+	}
+
+	return nil
+}
+
 type byteArrayType struct{}
 
 func (t byteArrayType) String() string                           { return "BYTE_ARRAY" }
@@ -519,6 +596,21 @@ func (t byteArrayType) Decode(dst encoding.Values, src []byte, enc encoding.Enco
 	return encoding.DecodeByteArray(dst, src, enc)
 }
 
+func (t byteArrayType) AssignValue(dst reflect.Value, src Value) error {
+	v := src.ByteArray()
+	switch dst.Kind() {
+	case reflect.String:
+		dst.SetString(string(v))
+	case reflect.Slice:
+		dst.SetBytes(copyBytes(v))
+	default:
+		val := reflect.ValueOf(string(v))
+		dst.Set(val)
+	}
+
+	return nil
+}
+
 type fixedLenByteArrayType struct{ length int }
 
 func (t fixedLenByteArrayType) String() string {
@@ -571,6 +663,31 @@ func (t fixedLenByteArrayType) Decode(dst encoding.Values, src []byte, enc encod
 	return encoding.DecodeFixedLenByteArray(dst, src, enc)
 }
 
+func (t fixedLenByteArrayType) AssignValue(dst reflect.Value, src Value) error {
+	v := src.ByteArray()
+	switch dst.Kind() {
+	case reflect.Array:
+		if dst.Type().Elem().Kind() == reflect.Uint8 && dst.Len() == len(v) {
+			// This code could be implemented as a call to reflect.Copy but
+			// it would require creating a reflect.Value from v which causes
+			// the heap allocation to pack the []byte value. To avoid this
+			// overhead we instead convert the reflect.Value holding the
+			// destination array into a byte slice which allows us to use
+			// a more efficient call to copy.
+			d := unsafe.Slice((*byte)(unsafecast.PointerOfValue(dst)), len(v))
+			copy(d, v)
+			return nil
+		}
+	case reflect.Slice:
+		dst.SetBytes(copyBytes(v))
+		return nil
+	}
+
+	val := reflect.ValueOf(copyBytes(v))
+	dst.Set(val)
+	return nil
+}
+
 // BE128 stands for "big-endian 128 bits". This type is used as a special case
 // for fixed-length byte arrays of 16 bytes, which are commonly used to
 // represent columns of random unique identifiers such as UUIDs.
@@ -632,6 +749,10 @@ func (t be128Type) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeFixedLenByteArray(dst, src, enc)
 }
 
+func (t be128Type) AssignValue(dst reflect.Value, src Value) error {
+	return fixedLenByteArrayType{length: 16}.AssignValue(dst, src)
+}
+
 // FixedLenByteArrayType constructs a type for fixed-length values of the given
 // size (in bytes).
 func FixedLenByteArrayType(length int) Type {
@@ -838,6 +959,14 @@ func (t *intType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding)
 	}
 }
 
+func (t *intType) AssignValue(dst reflect.Value, src Value) error {
+	if t.BitWidth == 64 {
+		return Int64Type.AssignValue(dst, src)
+	} else {
+		return Int32Type.AssignValue(dst, src)
+	}
+}
+
 // Decimal constructs a leaf node of decimal logical type with the given
 // scale, precision, and underlying type.
 //
@@ -935,6 +1064,10 @@ func (t *stringType) Decode(dst encoding.Values, src []byte, enc encoding.Encodi
 	return encoding.DecodeByteArray(dst, src, enc)
 }
 
+func (t *stringType) AssignValue(dst reflect.Value, src Value) error {
+	return ByteArrayType.AssignValue(dst, src)
+}
+
 // UUID constructs a leaf node of UUID logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#uuid
@@ -992,6 +1125,10 @@ func (t *uuidType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeFixedLenByteArray(dst, src, enc)
 }
 
+func (t *uuidType) AssignValue(dst reflect.Value, src Value) error {
+	return fixedLenByteArrayType{length: 16}.AssignValue(dst, src)
+}
+
 // Enum constructs a leaf node with a logical type representing enumerations.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#enum
@@ -1055,6 +1192,10 @@ func (t *enumType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeByteArray(dst, src, enc)
 }
 
+func (t *enumType) AssignValue(dst reflect.Value, src Value) error {
+	return ByteArrayType.AssignValue(dst, src)
+}
+
 // JSON constructs a leaf node of JSON logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#json
@@ -1118,6 +1259,10 @@ func (t *jsonType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeByteArray(dst, src, enc)
 }
 
+func (t *jsonType) AssignValue(dst reflect.Value, src Value) error {
+	return ByteArrayType.AssignValue(dst, src)
+}
+
 // BSON constructs a leaf node of BSON logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#bson
@@ -1181,6 +1326,10 @@ func (t *bsonType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeByteArray(dst, src, enc)
 }
 
+func (t *bsonType) AssignValue(dst reflect.Value, src Value) error {
+	return ByteArrayType.AssignValue(dst, src)
+}
+
 // Date constructs a leaf node of DATE logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#date
@@ -1240,6 +1389,10 @@ func (t *dateType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	return encoding.DecodeInt32(dst, src, enc)
 }
 
+func (t *dateType) AssignValue(dst reflect.Value, src Value) error {
+	return Int32Type.AssignValue(dst, src)
+}
+
 // TimeUnit represents units of time in the parquet type system.
 type TimeUnit interface {
 	// Returns the precision of the time unit as a time.Duration value.
@@ -1412,6 +1565,14 @@ func (t *timeType) Decode(dst encoding.Values, src []byte, enc encoding.Encoding
 	}
 }
 
+func (t *timeType) AssignValue(dst reflect.Value, src Value) error {
+	if t.useInt32() {
+		return Int32Type.AssignValue(dst, src)
+	} else {
+		return Int64Type.AssignValue(dst, src)
+	}
+}
+
 // Timestamp constructs of leaf node of TIMESTAMP logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp
@@ -1478,6 +1639,10 @@ func (t *timestampType) Decode(dst encoding.Values, src []byte, enc encoding.Enc
 	return encoding.DecodeInt64(dst, src, enc)
 }
 
+func (t *timestampType) AssignValue(dst reflect.Value, src Value) error {
+	return Int64Type.AssignValue(dst, src)
+}
+
 // List constructs a node of LIST logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#lists
@@ -1541,6 +1706,10 @@ func (t *listType) Decode(_ encoding.Values, _ []byte, _ encoding.Encoding) (enc
 	panic("cannot decode parquet LIST type")
 }
 
+func (t *listType) AssignValue(reflect.Value, Value) error {
+	panic("cannot assign value to a parquet LIST type")
+}
+
 // Map constructs a node of MAP logical type.
 //
 // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#maps
@@ -1609,6 +1778,10 @@ func (t *mapType) Decode(_ encoding.Values, _ []byte, _ encoding.Encoding) (enco
 	panic("cannot decode parquet MAP type")
 }
 
+func (t *mapType) AssignValue(reflect.Value, Value) error {
+	panic("cannot assign value to a parquet MAP type")
+}
+
 type nullType format.NullType
 
 func (t *nullType) String() string { return (*format.NullType)(t).String() }
@@ -1659,6 +1832,10 @@ func (t *nullType) Decode(dst encoding.Values, _ []byte, _ encoding.Encoding) (e
 	return dst, nil
 }
 
+func (t *nullType) AssignValue(reflect.Value, Value) error {
+	return nil
+}
+
 type groupType struct{}
 
 func (groupType) String() string { return "group" }
@@ -1699,6 +1876,10 @@ func (groupType) Decode(_ encoding.Values, _ []byte, _ encoding.Encoding) (encod
 	panic("cannot decode parquet group")
 }
 
+func (groupType) AssignValue(reflect.Value, Value) error {
+	panic("cannot assign value to a parquet group")
+}
+
 func (groupType) Length() int { return 0 }
 
 func (groupType) EstimateSize(int) int64 { return 0 }
diff --git a/value.go b/value.go
index 2d4b299..d3ece01 100644
--- a/value.go
+++ b/value.go
@@ -383,7 +383,14 @@ func (v Value) Int32() int32 { return int32(v.u64) }
 func (v Value) Int64() int64 { return int64(v.u64) }
 
 // Int96 returns v as a int96, assuming the underlying type is INT96.
-func (v Value) Int96() deprecated.Int96 { return makeInt96(v.ByteArray()) }
+func (v Value) Int96() deprecated.Int96 {
+	var val deprecated.Int96
+	if !v.IsNull() {
+		val = makeInt96(v.ByteArray())
+	}
+
+	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)) }
@@ -589,124 +596,6 @@ func makeInt96(bits []byte) (i96 deprecated.Int96) {
 	}
 }
 
-func assignValue(dst reflect.Value, src Value) error {
-	if src.IsNull() {
-		dst.Set(reflect.Zero(dst.Type()))
-		return nil
-	}
-
-	dstKind := dst.Kind()
-	srcKind := src.Kind()
-
-	var val reflect.Value
-	switch srcKind {
-	case Boolean:
-		v := src.Boolean()
-		switch dstKind {
-		case reflect.Bool:
-			dst.SetBool(v)
-			return nil
-		default:
-			val = reflect.ValueOf(v)
-		}
-
-	case Int32:
-		v := int64(src.Int32())
-		switch dstKind {
-		case reflect.Int8, reflect.Int16, reflect.Int32:
-			dst.SetInt(int64(v))
-			return nil
-		case reflect.Uint8, reflect.Uint16, reflect.Uint32:
-			dst.SetUint(uint64(v))
-			return nil
-		default:
-			val = reflect.ValueOf(v)
-		}
-
-	case Int64:
-		v := src.Int64()
-		switch dstKind {
-		case reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Int:
-			dst.SetInt(v)
-			return nil
-		case reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uint, reflect.Uintptr:
-			dst.SetUint(uint64(v))
-			return nil
-		default:
-			val = reflect.ValueOf(v)
-		}
-
-	case Int96:
-		val = reflect.ValueOf(src.Int96())
-
-	case Float:
-		v := src.Float()
-		switch dstKind {
-		case reflect.Float32, reflect.Float64:
-			dst.SetFloat(float64(v))
-			return nil
-		default:
-			val = reflect.ValueOf(v)
-		}
-
-	case Double:
-		v := src.Double()
-		switch dstKind {
-		case reflect.Float32, reflect.Float64:
-			dst.SetFloat(v)
-			return nil
-		default:
-			val = reflect.ValueOf(v)
-		}
-
-	case ByteArray:
-		v := src.ByteArray()
-		switch dstKind {
-		case reflect.String:
-			dst.SetString(string(v))
-			return nil
-		case reflect.Slice:
-			if dst.Type().Elem().Kind() == reflect.Uint8 {
-				dst.SetBytes(copyBytes(v))
-				return nil
-			}
-		default:
-			val = reflect.ValueOf(string(v))
-		}
-
-	case FixedLenByteArray:
-		v := src.ByteArray()
-		switch dstKind {
-		case reflect.Array:
-			if dst.Type().Elem().Kind() == reflect.Uint8 && dst.Len() == len(v) {
-				// This code could be implemented as a call to reflect.Copy but
-				// it would require creating a reflect.Value from v which causes
-				// the heap allocation to pack the []byte value. To avoid this
-				// overhead we instead convert the reflect.Value holding the
-				// destination array into a byte slice which allows us to use
-				// a more efficient call to copy.
-				d := unsafe.Slice((*byte)(unsafecast.PointerOfValue(dst)), len(v))
-				copy(d, v)
-				return nil
-			}
-		case reflect.Slice:
-			if dst.Type().Elem().Kind() == reflect.Uint8 {
-				dst.SetBytes(copyBytes(v))
-				return nil
-			}
-		default:
-			val = reflect.ValueOf(copyBytes(v))
-		}
-	}
-
-	if val.IsValid() && val.Type().AssignableTo(dst.Type()) {
-		dst.Set(val)
-		return nil
-	}
-
-	return fmt.Errorf("cannot assign parquet value of type %s to go value of type %s", srcKind.String(), dst.Type())
-}
-
 func parseValue(kind Kind, data []byte) (val Value, err error) {
 	switch kind {
 	case Boolean:
diff --git a/writer.go b/writer.go
index 1a1be8a..d9fbf18 100644
--- a/writer.go
+++ b/writer.go
@@ -211,9 +211,11 @@ func (w *Writer) ReadRowsFrom(rows RowReader) (written int64, err error) {
 func (w *Writer) Schema() *Schema { return w.schema }
 
 type writer struct {
-	buffer *bufio.Writer
-	writer offsetTrackingWriter
-	values [][]Value
+	buffer  *bufio.Writer
+	writer  offsetTrackingWriter
+	values  [][]Value
+	numRows int64
+	maxRows int64
 
 	createdBy string
 	metadata  []format.KeyValue
@@ -240,6 +242,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		w.buffer = bufio.NewWriterSize(output, config.WriteBufferSize)
 		w.writer.Reset(w.buffer)
 	}
+	w.maxRows = config.MaxRowsPerRowGroup
 	w.createdBy = config.CreatedBy
 	w.metadata = make([]format.KeyValue, 0, len(config.KeyValueMetadata))
 	for k, v := range config.KeyValueMetadata {
@@ -550,6 +553,7 @@ func (w *writer) writeRowGroup(rowGroupSchema *Schema, rowGroupSortingColumns []
 	}
 
 	defer func() {
+		w.numRows = 0
 		for _, c := range w.columns {
 			c.reset()
 		}
@@ -653,33 +657,65 @@ func (w *writer) writeRowGroup(rowGroupSchema *Schema, rowGroupSortingColumns []
 }
 
 func (w *writer) WriteRows(rows []Row) (int, error) {
-	defer func() {
-		for i, values := range w.values {
-			clearValues(values)
-			w.values[i] = values[:0]
+	return w.writeRows(len(rows), func(start, end int) (int, error) {
+		defer func() {
+			for i, values := range w.values {
+				clearValues(values)
+				w.values[i] = values[:0]
+			}
+		}()
+
+		// TODO: if an error occurs in this method the writer may be left in an
+		// partially functional state. Applications are not expected to continue
+		// using the writer after getting an error, but maybe we could ensure that
+		// we are preventing further use as well?
+		for _, row := range rows[start:end] {
+			for _, value := range row {
+				columnIndex := value.Column()
+				w.values[columnIndex] = append(w.values[columnIndex], value)
+			}
 		}
-	}()
 
-	// TODO: if an error occurs in this method the writer may be left in an
-	// partially functional state. Applications are not expected to continue
-	// using the writer after getting an error, but maybe we could ensure that
-	// we are preventing further use as well?
-	for _, row := range rows {
-		for _, value := range row {
-			columnIndex := value.Column()
-			w.values[columnIndex] = append(w.values[columnIndex], value)
+		for i, values := range w.values {
+			if len(values) > 0 {
+				if err := w.columns[i].writeRows(values); err != nil {
+					return 0, err
+				}
+			}
 		}
-	}
 
-	for i, values := range w.values {
-		if len(values) > 0 {
-			if err := w.columns[i].writeRows(values); err != nil {
-				return 0, err
+		return end - start, nil
+	})
+}
+
+func (w *writer) writeRows(numRows int, write func(i, j int) (int, error)) (int, error) {
+	written := 0
+
+	for written < numRows {
+		remain := w.maxRows - w.numRows
+		length := numRows - written
+
+		if remain == 0 {
+			remain = w.maxRows
+
+			if err := w.flush(); err != nil {
+				return written, err
 			}
 		}
+
+		if remain < int64(length) {
+			length = int(remain)
+		}
+
+		n, err := write(written, written+length)
+		written += n
+		w.numRows += int64(n)
+		if err != nil {
+			return written, err
+		}
 	}
 
-	return len(rows), nil
+	return written, nil
 }
 
 // The WriteValues method is intended to work in pair with WritePage to allow
diff --git a/writer_go18.go b/writer_go18.go
index 92e0d0e..1145894 100644
--- a/writer_go18.go
+++ b/writer_go18.go
@@ -148,22 +148,24 @@ func (w *GenericWriter[T]) Reset(output io.Writer) {
 }
 
 func (w *GenericWriter[T]) Write(rows []T) (int, error) {
-	n, err := w.write(w, rows)
-	if err != nil {
-		return n, err
-	}
+	return w.base.writer.writeRows(len(rows), func(i, j int) (int, error) {
+		n, err := w.write(w, rows[i:j:j])
+		if err != nil {
+			return n, err
+		}
 
-	for _, c := range w.base.writer.columns {
-		c.numValues = int32(c.columnBuffer.NumValues())
+		for _, c := range w.base.writer.columns {
+			c.numValues = int32(c.columnBuffer.NumValues())
 
-		if c.numValues > 0 && c.numValues >= c.maxValues {
-			if err := c.flush(); err != nil {
-				return 0, err
+			if c.numValues > 0 && c.numValues >= c.maxValues {
+				if err := c.flush(); err != nil {
+					return 0, err
+				}
 			}
 		}
-	}
 
-	return n, nil
+		return n, nil
+	})
 }
 
 func (w *GenericWriter[T]) WriteRows(rows []Row) (int, error) {


Test patch:
diff --git a/encoding/bytestreamsplit/bytestreamsplit.go b/encoding/bytestreamsplit/bytestreamsplit.go
index 5f7899c..bd2525a 100644
--- a/encoding/bytestreamsplit/bytestreamsplit.go
+++ b/encoding/bytestreamsplit/bytestreamsplit.go
@@ -52,7 +52,7 @@ func (e *Encoding) DecodeDouble(dst []float64, src []byte) ([]float64, error) {
 
 func resize(buf []byte, size int) []byte {
 	if cap(buf) < size {
-		buf = make([]byte, size)
+		buf = make([]byte, size, 2*size)
 	} else {
 		buf = buf[:size]
 	}
diff --git a/search_test.go b/search_test.go
index 76f40ac..e10fd99 100644
--- a/search_test.go
+++ b/search_test.go
@@ -1,8 +1,6 @@
 package parquet_test
 
 import (
-	"fmt"
-	"math"
 	"testing"
 
 	"github.com/segmentio/parquet-go"
@@ -34,11 +32,24 @@ func TestSearchBinary(t *testing.T) {
 	testSearch(t, [][]int32{
 		{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
 		{10, 10, 10, 10},
-		{21, 22, 23, 24, 25},
-		{30},
-		{31},
+		{21, 22, 24, 25, 30},
+		{30, 30},
+		{30, 31},
 		{32},
 		{42, 43, 44, 45, 46, 47, 48, 49},
+	}, [][]int{
+		{10, 1},
+		{0, 0},
+		{9, 0},
+		// non-existant, but would be in this page
+		{23, 2},
+		// ensure we find the first page
+		{30, 2},
+		{31, 4},
+		// out of bounds
+		{99, 7},
+		// out of bounds
+		{-1, 7},
 	})
 }
 
@@ -47,12 +58,23 @@ func TestSearchLinear(t *testing.T) {
 		{10, 10, 10, 10},
 		{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
 		{21, 22, 23, 24, 25},
-		{19, 18, 17, 16, 15, 14, 13, 12, 11},
+		{19, 18, 17, 16, 14, 13, 12, 11},
 		{42, 43, 44, 45, 46, 47, 48, 49},
+	}, [][]int{
+		{10, 0},
+		{0, 1},
+		{9, 1},
+		{48, 4},
+		// non-existant, but could be in this page
+		{15, 3},
+		// out of bounds
+		{99, 5},
+		// out of bounds
+		{-1, 5},
 	})
 }
 
-func testSearch(t *testing.T, pages [][]int32) {
+func testSearch(t *testing.T, pages [][]int32, expectIndex [][]int) {
 	indexer := parquet.Int32Type.NewColumnIndexer(0)
 
 	for _, values := range pages {
@@ -77,22 +99,12 @@ func testSearch(t *testing.T, pages [][]int32) {
 	formatIndex := indexer.ColumnIndex()
 	columnIndex := parquet.NewColumnIndex(parquet.Int32, &formatIndex)
 
-	for i, values := range pages {
-		t.Run(fmt.Sprintf("page#%02d", i), func(t *testing.T) {
-			for _, value := range values {
-				v := parquet.ValueOf(value)
-				j := parquet.Search(columnIndex, v, parquet.Int32Type)
-
-				if i != j {
-					t.Errorf("searching for value %v: got=%d want=%d", v, j, i)
-				}
-			}
+	for _, values := range expectIndex {
+		v := parquet.ValueOf(values[0])
+		j := parquet.Search(columnIndex, v, parquet.Int32Type)
 
-			for _, test := range []int32{math.MinInt32, math.MaxInt32} {
-				if page := parquet.Search(columnIndex, parquet.ValueOf(test), parquet.Int32Type); page != len(pages) {
-					t.Errorf("search for non-existing value %v: got=%d want=%d", test, page, len(pages))
-				}
-			}
-		})
+		if values[1] != j {
+			t.Errorf("searching for value %v: got=%d want=%d", v, j, values[1])
+		}
 	}
 }
diff --git a/value_test.go b/value_test.go
index cf2fa05..db9d7dd 100644
--- a/value_test.go
+++ b/value_test.go
@@ -1,11 +1,13 @@
 package parquet_test
 
 import (
+	"bytes"
 	"math"
 	"testing"
 	"unsafe"
 
 	"github.com/segmentio/parquet-go"
+	"github.com/segmentio/parquet-go/deprecated"
 )
 
 func TestSizeOfValue(t *testing.T) {
@@ -90,3 +92,52 @@ func TestValueClone(t *testing.T) {
 		})
 	}
 }
+
+func TestZeroValue(t *testing.T) {
+	var v parquet.Value
+	if !v.IsNull() {
+		t.Error("expected zero value parquet.Value to be null")
+	}
+
+	if v.Byte() != byte(0) {
+		t.Errorf("byte not zero value: got=%#v", v.Byte())
+	}
+
+	if v.Boolean() != false {
+		t.Errorf("boolean not zero value: got=%#v", v.Boolean())
+	}
+
+	if v.Int32() != 0 {
+		t.Errorf("int32 not zero value: got=%#v", v.Int32())
+	}
+
+	if v.Int64() != 0 {
+		t.Errorf("int64 not zero value: got=%#v", v.Int64())
+	}
+
+	var zeroInt96 deprecated.Int96
+	if v.Int96() != zeroInt96 {
+		t.Errorf("int96 not zero value: got=%#v", zeroInt96)
+	}
+
+	if v.Float() != 0 {
+		t.Errorf("float not zero value: got=%#v", v.Float())
+	}
+
+	if v.Double() != 0 {
+		t.Errorf("double not zero value: got=%#v", v.Double())
+	}
+
+	if v.Uint32() != 0 {
+		t.Errorf("uint32 not zero value: got=%#v", v.Uint32())
+	}
+
+	if v.Uint64() != 0 {
+		t.Errorf("uint64 not zero value: got=%#v", v.Uint64())
+	}
+
+	var zeroByte []byte
+	if !bytes.Equal(v.ByteArray(), zeroByte) {
+		t.Errorf("byte array not zero value: got=%#v", v.ByteArray())
+	}
+}
diff --git a/writer_go18_test.go b/writer_go18_test.go
index c26cb97..66f1b53 100644
--- a/writer_go18_test.go
+++ b/writer_go18_test.go
@@ -315,3 +315,40 @@ func TestIssue347Writer(t *testing.T) {
 	}()
 	_ = parquet.NewGenericWriter[any](b)
 }
+
+func TestIssue375(t *testing.T) {
+	type Row struct{ FirstName, LastName string }
+
+	output := new(bytes.Buffer)
+	writer := parquet.NewGenericWriter[Row](output, parquet.MaxRowsPerRowGroup(10))
+
+	rows := make([]Row, 100)
+	for i := range rows {
+		rows[i] = Row{
+			FirstName: "0123456789"[i%10 : i%10+1],
+			LastName:  "foo",
+		}
+	}
+
+	n, err := writer.Write(rows)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != len(rows) {
+		t.Fatal("wrong number of rows written:", n)
+	}
+
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	f, err := parquet.OpenFile(bytes.NewReader(output.Bytes()), int64(output.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rowGroups := f.RowGroups()
+	if len(rowGroups) != 10 {
+		t.Errorf("wrong number of row groups in parquet file: want=10 got=%d", len(rowGroups))
+	}
+}
diff --git a/writer_test.go b/writer_test.go
index cb54e35..4ebe7ea 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -682,3 +682,32 @@ func TestWriterResetWithBloomFilters(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestWriterMaxRowsPerRowGroup(t *testing.T) {
+	output := new(bytes.Buffer)
+	writer := parquet.NewWriter(output, parquet.MaxRowsPerRowGroup(10))
+
+	for i := 0; i < 100; i++ {
+		err := writer.Write(struct{ FirstName, LastName string }{
+			FirstName: "0123456789"[i%10 : i%10+1],
+			LastName:  "foo",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	f, err := parquet.OpenFile(bytes.NewReader(output.Bytes()), int64(output.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rowGroups := f.RowGroups()
+	if len(rowGroups) != 10 {
+		t.Errorf("wrong number of row groups in parquet file: want=10 got=%d", len(rowGroups))
+	}
+}