parquet_generated_test.go

package parquet_test

// Code generated by github.com/parsyl/parquet.  DO NOT EDIT.

import (
	"encoding/binary"
	"fmt"
	"io"
	"math"
	"strings"

	"github.com/parsyl/parquet"
	sch "github.com/parsyl/parquet/schema"
	"github.com/valyala/bytebufferpool"
)

var _ = math.MaxInt32 // to avoid unused import

type compression int

const (
	compressionUncompressed compression = 0
	compressionSnappy       compression = 1
	compressionGzip         compression = 2
	compressionUnknown      compression = -1
)

var buffpool = bytebufferpool.Pool{}

// ParquetWriter reprents a row group
type ParquetWriter struct {
	fields []Field

	len int

	// child points to the next page
	child *ParquetWriter

	// max is the number of Record items that can get written before
	// a new set of column chunks is written
	max int

	meta        *parquet.Metadata
	w           io.Writer
	compression compression
}

func Fields(compression compression) []Field {
	return []Field{
		NewInt32Field(readID, writeID, []string{"id"}, fieldCompression(compression)),
		NewStringField(readName, writeName, []string{"name"}, fieldCompression(compression)),
		NewInt32OptionalField(readAge, writeAge, []string{"age"}, []int{1}, optionalFieldCompression(compression)),
		NewInt64Field(readHappiness, writeHappiness, []string{"happiness"}, fieldCompression(compression)),
		NewInt64OptionalField(readSadness, writeSadness, []string{"sadness"}, []int{1}, optionalFieldCompression(compression)),
		NewStringOptionalField(readCode, writeCode, []string{"code"}, []int{1}, optionalFieldCompression(compression)),
		NewFloat32Field(readFunkiness, writeFunkiness, []string{"funkiness"}, fieldCompression(compression)),
		NewFloat64Field(readBoldness, writeBoldness, []string{"boldness"}, fieldCompression(compression)),
		NewFloat32OptionalField(readLameness, writeLameness, []string{"lameness"}, []int{1}, optionalFieldCompression(compression)),
		NewBoolOptionalField(readKeen, writeKeen, []string{"keen"}, []int{1}, optionalFieldCompression(compression)),
		NewUint32Field(readBirthday, writeBirthday, []string{"birthday"}, fieldCompression(compression)),
		NewUint64OptionalField(readAnniversary, writeAnniversary, []string{"anniversary"}, []int{1}, optionalFieldCompression(compression)),
		NewStringField(readBFF, writeBFF, []string{"bff"}, fieldCompression(compression)),
		NewBoolField(readHungry, writeHungry, []string{"hungry"}, fieldCompression(compression)),
		NewStringOptionalField(readHobbyName, writeHobbyName, []string{"hobby", "name"}, []int{1, 0}, optionalFieldCompression(compression)),
		NewInt32OptionalField(readHobbyDifficulty, writeHobbyDifficulty, []string{"hobby", "difficulty"}, []int{1, 1}, optionalFieldCompression(compression)),
		NewStringOptionalField(readHobbySkillsName, writeHobbySkillsName, []string{"hobby", "skills", "name"}, []int{1, 2, 0}, optionalFieldCompression(compression)),
		NewStringOptionalField(readHobbySkillsDifficulty, writeHobbySkillsDifficulty, []string{"hobby", "skills", "difficulty"}, []int{1, 2, 0}, optionalFieldCompression(compression)),
		NewInt32OptionalField(readFriendsID, writeFriendsID, []string{"friends", "id"}, []int{2, 0}, optionalFieldCompression(compression)),
		NewStringOptionalField(readFriendsName, writeFriendsName, []string{"friends", "name"}, []int{2, 0}, optionalFieldCompression(compression)),
		NewInt32OptionalField(readFriendsAge, writeFriendsAge, []string{"friends", "age"}, []int{2, 1}, optionalFieldCompression(compression)),
		NewBoolField(readSleepy, writeSleepy, []string{"Sleepy"}, fieldCompression(compression)),
	}
}

func readID(x Person) int32 {
	return x.ID
}

func writeID(x *Person, vals []int32) {
	x.ID = vals[0]
}

func readName(x Person) string {
	return x.Name
}

func writeName(x *Person, vals []string) {
	x.Name = vals[0]
}

func readAge(x Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8) {
	switch {
	case x.Age == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Age)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeAge(x *Person, vals []int32, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Age = pint32(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readHappiness(x Person) int64 {
	return x.Happiness
}

func writeHappiness(x *Person, vals []int64) {
	x.Happiness = vals[0]
}

func readSadness(x Person, vals []int64, defs, reps []uint8) ([]int64, []uint8, []uint8) {
	switch {
	case x.Sadness == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Sadness)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeSadness(x *Person, vals []int64, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Sadness = pint64(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readCode(x Person, vals []string, defs, reps []uint8) ([]string, []uint8, []uint8) {
	switch {
	case x.Code == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Code)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeCode(x *Person, vals []string, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Code = pstring(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readFunkiness(x Person) float32 {
	return x.Funkiness
}

func writeFunkiness(x *Person, vals []float32) {
	x.Funkiness = vals[0]
}

func readBoldness(x Person) float64 {
	return x.Boldness
}

func writeBoldness(x *Person, vals []float64) {
	x.Boldness = vals[0]
}

func readLameness(x Person, vals []float32, defs, reps []uint8) ([]float32, []uint8, []uint8) {
	switch {
	case x.Lameness == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Lameness)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeLameness(x *Person, vals []float32, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Lameness = pfloat32(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readKeen(x Person, vals []bool, defs, reps []uint8) ([]bool, []uint8, []uint8) {
	switch {
	case x.Keen == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Keen)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeKeen(x *Person, vals []bool, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Keen = pbool(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readBirthday(x Person) uint32 {
	return x.Birthday
}

func writeBirthday(x *Person, vals []uint32) {
	x.Birthday = vals[0]
}

func readAnniversary(x Person, vals []uint64, defs, reps []uint8) ([]uint64, []uint8, []uint8) {
	switch {
	case x.Anniversary == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Anniversary)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeAnniversary(x *Person, vals []uint64, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Anniversary = puint64(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readBFF(x Person) string {
	return x.BFF
}

func writeBFF(x *Person, vals []string) {
	x.BFF = vals[0]
}

func readHungry(x Person) bool {
	return x.Hungry
}

func writeHungry(x *Person, vals []bool) {
	x.Hungry = vals[0]
}

func readHobbyName(x Person, vals []string, defs, reps []uint8) ([]string, []uint8, []uint8) {
	switch {
	case x.Hobby == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	default:
		vals = append(vals, x.Hobby.Name)
		defs = append(defs, 1)
		return vals, defs, reps
	}
}

func writeHobbyName(x *Person, vals []string, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 1:
		x.Hobby = &Hobby{Name: vals[0]}
		return 1, 1
	}

	return 0, 1
}

func readHobbyDifficulty(x Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8) {
	switch {
	case x.Hobby == nil:
		defs = append(defs, 0)
		return vals, defs, reps
	case x.Hobby.Difficulty == nil:
		defs = append(defs, 1)
		return vals, defs, reps
	default:
		vals = append(vals, *x.Hobby.Difficulty)
		defs = append(defs, 2)
		return vals, defs, reps
	}
}

func writeHobbyDifficulty(x *Person, vals []int32, defs, reps []uint8) (int, int) {
	def := defs[0]
	switch def {
	case 2:
		x.Hobby.Difficulty = pint32(vals[0])
		return 1, 1
	}

	return 0, 1
}

func readHobbySkillsName(x Person, vals []string, defs, reps []uint8) ([]string, []uint8, []uint8) {
	var lastRep uint8

	if x.Hobby == nil {
		defs = append(defs, 0)
		reps = append(reps, lastRep)
	} else {
		if len(x.Hobby.Skills) == 0 {
			defs = append(defs, 1)
			reps = append(reps, lastRep)
		} else {
			for i0, x0 := range x.Hobby.Skills {
				if i0 >= 1 {
					lastRep = 1
				}
				defs = append(defs, 2)
				reps = append(reps, lastRep)
				vals = append(vals, x0.Name)
			}
		}
	}

	return vals, defs, reps
}

func writeHobbySkillsName(x *Person, vals []string, defs, reps []uint8) (int, int) {
	var nVals, nLevels int
	ind := make(indices, 1)

	for i := range defs {
		def := defs[i]
		rep := reps[i]
		if i > 0 && rep == 0 {
			break
		}

		nLevels++
		ind.rep(rep)

		switch def {
		case 2:
			x.Hobby.Skills = append(x.Hobby.Skills, Skill{Name: vals[nVals]})
			nVals++
		}
	}

	return nVals, nLevels
}

func readHobbySkillsDifficulty(x Person, vals []string, defs, reps []uint8) ([]string, []uint8, []uint8) {
	var lastRep uint8

	if x.Hobby == nil {
		defs = append(defs, 0)
		reps = append(reps, lastRep)
	} else {
		if len(x.Hobby.Skills) == 0 {
			defs = append(defs, 1)
			reps = append(reps, lastRep)
		} else {
			for i0, x0 := range x.Hobby.Skills {
				if i0 >= 1 {
					lastRep = 1
				}
				defs = append(defs, 2)
				reps = append(reps, lastRep)
				vals = append(vals, x0.Difficulty)
			}
		}
	}

	return vals, defs, reps
}

func writeHobbySkillsDifficulty(x *Person, vals []string, defs, reps []uint8) (int, int) {
	var nVals, nLevels int
	ind := make(indices, 1)

	for i := range defs {
		def := defs[i]
		rep := reps[i]
		if i > 0 && rep == 0 {
			break
		}

		nLevels++
		ind.rep(rep)

		switch def {
		case 2:
			x.Hobby.Skills[ind[0]].Difficulty = vals[nVals]
			nVals++
		}
	}

	return nVals, nLevels
}

func readFriendsID(x Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8) {
	var lastRep uint8

	if len(x.Friends) == 0 {
		defs = append(defs, 0)
		reps = append(reps, lastRep)
	} else {
		for i0, x0 := range x.Friends {
			if i0 >= 1 {
				lastRep = 1
			}
			defs = append(defs, 1)
			reps = append(reps, lastRep)
			vals = append(vals, x0.ID)
		}
	}

	return vals, defs, reps
}

func writeFriendsID(x *Person, vals []int32, defs, reps []uint8) (int, int) {
	var nVals, nLevels int
	ind := make(indices, 1)

	for i := range defs {
		def := defs[i]
		rep := reps[i]
		if i > 0 && rep == 0 {
			break
		}

		nLevels++
		ind.rep(rep)

		switch def {
		case 1:
			x.Friends = append(x.Friends, Being{ID: vals[nVals]})
			nVals++
		}
	}

	return nVals, nLevels
}

func readFriendsName(x Person, vals []string, defs, reps []uint8) ([]string, []uint8, []uint8) {
	var lastRep uint8

	if len(x.Friends) == 0 {
		defs = append(defs, 0)
		reps = append(reps, lastRep)
	} else {
		for i0, x0 := range x.Friends {
			if i0 >= 1 {
				lastRep = 1
			}
			defs = append(defs, 1)
			reps = append(reps, lastRep)
			vals = append(vals, x0.Name)
		}
	}

	return vals, defs, reps
}

func writeFriendsName(x *Person, vals []string, defs, reps []uint8) (int, int) {
	var nVals, nLevels int
	ind := make(indices, 1)

	for i := range defs {
		def := defs[i]
		rep := reps[i]
		if i > 0 && rep == 0 {
			break
		}

		nLevels++
		ind.rep(rep)

		switch def {
		case 1:
			x.Friends[ind[0]].Name = vals[nVals]
			nVals++
		}
	}

	return nVals, nLevels
}

func readFriendsAge(x Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8) {
	var lastRep uint8

	if len(x.Friends) == 0 {
		defs = append(defs, 0)
		reps = append(reps, lastRep)
	} else {
		for i0, x0 := range x.Friends {
			if i0 >= 1 {
				lastRep = 1
			}
			if x0.Age == nil {
				defs = append(defs, 1)
				reps = append(reps, lastRep)
			} else {
				defs = append(defs, 2)
				reps = append(reps, lastRep)
				vals = append(vals, *x0.Age)
			}
		}
	}

	return vals, defs, reps
}

func writeFriendsAge(x *Person, vals []int32, defs, reps []uint8) (int, int) {
	var nVals, nLevels int
	ind := make(indices, 1)

	for i := range defs {
		def := defs[i]
		rep := reps[i]
		if i > 0 && rep == 0 {
			break
		}

		nLevels++
		ind.rep(rep)

		switch def {
		case 2:
			x.Friends[ind[0]].Age = pint32(vals[nVals])
			nVals++
		}
	}

	return nVals, nLevels
}

func readSleepy(x Person) bool {
	return x.Sleepy
}

func writeSleepy(x *Person, vals []bool) {
	x.Sleepy = vals[0]
}

func fieldCompression(c compression) func(*parquet.RequiredField) {
	switch c {
	case compressionUncompressed:
		return parquet.RequiredFieldUncompressed
	case compressionSnappy:
		return parquet.RequiredFieldSnappy
	case compressionGzip:
		return parquet.RequiredFieldGzip
	default:
		return parquet.RequiredFieldUncompressed
	}
}

func optionalFieldCompression(c compression) func(*parquet.OptionalField) {
	switch c {
	case compressionUncompressed:
		return parquet.OptionalFieldUncompressed
	case compressionSnappy:
		return parquet.OptionalFieldSnappy
	case compressionGzip:
		return parquet.OptionalFieldGzip
	default:
		return parquet.OptionalFieldUncompressed
	}
}

func NewParquetWriter(w io.Writer, opts ...func(*ParquetWriter) error) (*ParquetWriter, error) {
	return newParquetWriter(w, append(opts, begin)...)
}

func newParquetWriter(w io.Writer, opts ...func(*ParquetWriter) error) (*ParquetWriter, error) {
	p := &ParquetWriter{
		max:         1000,
		w:           w,
		compression: compressionSnappy,
	}

	for _, opt := range opts {
		if err := opt(p); err != nil {
			return nil, err
		}
	}

	p.fields = Fields(p.compression)
	if p.meta == nil {
		ff := Fields(p.compression)
		schema := make([]parquet.Field, len(ff))
		for i, f := range ff {
			schema[i] = f.Schema()
		}
		p.meta = parquet.New(schema...)
	}

	return p, nil
}

// MaxPageSize is the maximum number of rows in each row groups' page.
func MaxPageSize(m int) func(*ParquetWriter) error {
	return func(p *ParquetWriter) error {
		p.max = m
		return nil
	}
}

var par1 = []byte("PAR1")

func begin(p *ParquetWriter) error {
	_, err := p.w.Write(par1)
	return err
}

func withMeta(m *parquet.Metadata) func(*ParquetWriter) error {
	return func(p *ParquetWriter) error {
		p.meta = m
		return nil
	}
}

func Uncompressed(p *ParquetWriter) error {
	p.compression = compressionUncompressed
	return nil
}

func Snappy(p *ParquetWriter) error {
	p.compression = compressionSnappy
	return nil
}

func Gzip(p *ParquetWriter) error {
	p.compression = compressionGzip
	return nil
}

func withCompression(c compression) func(*ParquetWriter) error {
	return func(p *ParquetWriter) error {
		p.compression = c
		return nil
	}
}

func (p *ParquetWriter) Write() error {
	for i, f := range p.fields {
		if err := f.Write(p.w, p.meta); err != nil {
			return err
		}

		for child := p.child; child != nil; child = child.child {
			if err := child.fields[i].Write(p.w, p.meta); err != nil {
				return err
			}
		}
	}

	p.fields = Fields(p.compression)
	p.child = nil
	p.len = 0

	schema := make([]parquet.Field, len(p.fields))
	for i, f := range p.fields {
		schema[i] = f.Schema()
	}
	p.meta.StartRowGroup(schema...)
	return nil
}

func (p *ParquetWriter) Close() error {
	if err := p.meta.Footer(p.w); err != nil {
		return err
	}

	_, err := p.w.Write(par1)
	return err
}

func (p *ParquetWriter) Add(rec Person) {
	if p.len == p.max {
		if p.child == nil {
			// an error can't happen here
			p.child, _ = newParquetWriter(p.w, MaxPageSize(p.max), withMeta(p.meta), withCompression(p.compression))
		}

		p.child.Add(rec)
		return
	}

	p.meta.NextDoc()
	for _, f := range p.fields {
		f.Add(rec)
	}

	p.len++
}

type Field interface {
	Add(r Person)
	Write(w io.Writer, meta *parquet.Metadata) error
	Schema() parquet.Field
	Scan(r *Person)
	Read(r io.ReadSeeker, pg parquet.Page) error
	Name() string
	Levels() ([]uint8, []uint8)
}

func getFields(ff []Field) map[string]Field {
	m := make(map[string]Field, len(ff))
	for _, f := range ff {
		m[f.Name()] = f
	}
	return m
}

func NewParquetReader(r io.ReadSeeker, opts ...func(*ParquetReader)) (*ParquetReader, error) {
	ff := Fields(compressionUnknown)
	pr := &ParquetReader{
		r: r,
	}

	for _, opt := range opts {
		opt(pr)
	}

	schema := make([]parquet.Field, len(ff))
	for i, f := range ff {
		pr.fieldNames = append(pr.fieldNames, f.Name())
		schema[i] = f.Schema()
	}

	meta := parquet.New(schema...)
	if err := meta.ReadFooter(r); err != nil {
		return nil, err
	}
	pr.rows = meta.Rows()
	var err error
	pr.pages, err = meta.Pages()
	if err != nil {
		return nil, err
	}

	pr.rowGroups = meta.RowGroups()
	_, err = r.Seek(4, io.SeekStart)
	if err != nil {
		return nil, err
	}
	pr.meta = meta

	return pr, pr.readRowGroup()
}

func readerIndex(i int) func(*ParquetReader) {
	return func(p *ParquetReader) {
		p.index = i
	}
}

// ParquetReader reads one page from a row group.
type ParquetReader struct {
	fields         map[string]Field
	fieldNames     []string
	index          int
	cursor         int64
	rows           int64
	rowGroupCursor int64
	rowGroupCount  int64
	pages          map[string][]parquet.Page
	meta           *parquet.Metadata
	err            error

	r         io.ReadSeeker
	rowGroups []parquet.RowGroup
}

type Levels struct {
	Name string
	Defs []uint8
	Reps []uint8
}

func (p *ParquetReader) Levels() []Levels {
	var out []Levels
	//for {
	for _, name := range p.fieldNames {
		f := p.fields[name]
		d, r := f.Levels()
		out = append(out, Levels{Name: f.Name(), Defs: d, Reps: r})
	}
	//	if err := p.readRowGroup(); err != nil {
	//		break
	//	}
	//}
	return out
}

func (p *ParquetReader) Error() error {
	return p.err
}

func (p *ParquetReader) readRowGroup() error {
	p.rowGroupCursor = 0

	if len(p.rowGroups) == 0 {
		p.rowGroupCount = 0
		return nil
	}

	rg := p.rowGroups[0]
	p.fields = getFields(Fields(compressionUnknown))
	p.rowGroupCount = rg.Rows
	p.rowGroupCursor = 0
	for _, col := range rg.Columns() {
		name := strings.Join(col.MetaData.PathInSchema, ".")
		f, ok := p.fields[name]
		if !ok {
			return fmt.Errorf("unknown field: %s", name)
		}
		pages := p.pages[name]
		if len(pages) <= p.index {
			break
		}

		pg := pages[0]
		if err := f.Read(p.r, pg); err != nil {
			return fmt.Errorf("unable to read field %s, err: %s", f.Name(), err)
		}
		p.pages[name] = p.pages[name][1:]
	}
	p.rowGroups = p.rowGroups[1:]
	return nil
}

func (p *ParquetReader) Rows() int64 {
	return p.rows
}

func (p *ParquetReader) Next() bool {
	if p.err == nil && p.cursor >= p.rows {
		return false
	}
	if p.rowGroupCursor >= p.rowGroupCount {
		p.err = p.readRowGroup()
		if p.err != nil {
			return false
		}
	}

	p.cursor++
	p.rowGroupCursor++
	return true
}

func (p *ParquetReader) Scan(x *Person) {
	if p.err != nil {
		return
	}

	for _, name := range p.fieldNames {
		f := p.fields[name]
		f.Scan(x)
	}
}

type Int32Field struct {
	vals []int32
	parquet.RequiredField
	read  func(r Person) int32
	write func(r *Person, vals []int32)
	stats *int32stats
}

func NewInt32Field(read func(r Person) int32, write func(r *Person, vals []int32), path []string, opts ...func(*parquet.RequiredField)) *Int32Field {
	return &Int32Field{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newInt32stats(),
	}
}

func (f *Int32Field) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Int32Type, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *Int32Field) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]int32, int(pg.N))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Int32Field) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint32(bs, uint32(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *Int32Field) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *Int32Field) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *Int32Field) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type StringField struct {
	parquet.RequiredField
	vals  []string
	read  func(r Person) string
	write func(r *Person, vals []string)
	stats *stringStats
}

func NewStringField(read func(r Person) string, write func(r *Person, vals []string), path []string, opts ...func(*parquet.RequiredField)) *StringField {
	return &StringField{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newStringStats(),
	}
}

func (f *StringField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: StringType, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *StringField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, s := range f.vals {
		binary.LittleEndian.PutUint32(bs, uint32(len(s)))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
		buf.WriteString(s)
	}

	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *StringField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	for j := 0; j < pg.N; j++ {
		var x int32
		if err := binary.Read(rr, binary.LittleEndian, &x); err != nil {
			return err
		}
		s := make([]byte, x)
		if _, err := rr.Read(s); err != nil {
			return err
		}

		f.vals = append(f.vals, string(s))
	}
	return nil
}

func (f *StringField) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *StringField) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *StringField) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type Int32OptionalField struct {
	parquet.OptionalField
	vals  []int32
	read  func(r Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8)
	write func(r *Person, vals []int32, defs, reps []uint8) (int, int)
	stats *int32optionalStats
}

func NewInt32OptionalField(read func(r Person, vals []int32, defs, reps []uint8) ([]int32, []uint8, []uint8), write func(r *Person, vals []int32, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *Int32OptionalField {
	return &Int32OptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newint32optionalStats(maxDef(types)),
	}
}

func (f *Int32OptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Int32Type, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *Int32OptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint32(bs, uint32(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.Defs), f.stats)
}

func (f *Int32OptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]int32, f.Values()-len(f.vals))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Int32OptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *Int32OptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *Int32OptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type Int64Field struct {
	vals []int64
	parquet.RequiredField
	read  func(r Person) int64
	write func(r *Person, vals []int64)
	stats *int64stats
}

func NewInt64Field(read func(r Person) int64, write func(r *Person, vals []int64), path []string, opts ...func(*parquet.RequiredField)) *Int64Field {
	return &Int64Field{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newInt64stats(),
	}
}

func (f *Int64Field) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Int64Type, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *Int64Field) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]int64, int(pg.N))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Int64Field) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 8)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint64(bs, uint64(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *Int64Field) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *Int64Field) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *Int64Field) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type Int64OptionalField struct {
	parquet.OptionalField
	vals  []int64
	read  func(r Person, vals []int64, defs, reps []uint8) ([]int64, []uint8, []uint8)
	write func(r *Person, vals []int64, defs, reps []uint8) (int, int)
	stats *int64optionalStats
}

func NewInt64OptionalField(read func(r Person, vals []int64, defs, reps []uint8) ([]int64, []uint8, []uint8), write func(r *Person, vals []int64, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *Int64OptionalField {
	return &Int64OptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newint64optionalStats(maxDef(types)),
	}
}

func (f *Int64OptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Int64Type, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *Int64OptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 8)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint64(bs, uint64(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.Defs), f.stats)
}

func (f *Int64OptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]int64, f.Values()-len(f.vals))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Int64OptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *Int64OptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *Int64OptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type StringOptionalField struct {
	parquet.OptionalField
	vals  []string
	read  func(r Person, vals []string, def, rep []uint8) ([]string, []uint8, []uint8)
	write func(r *Person, vals []string, def, rep []uint8) (int, int)
	stats *stringOptionalStats
}

func NewStringOptionalField(read func(r Person, vals []string, def, rep []uint8) ([]string, []uint8, []uint8), write func(r *Person, vals []string, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *StringOptionalField {
	return &StringOptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newStringOptionalStats(maxDef(types)),
	}
}

func (f *StringOptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: StringType, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *StringOptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *StringOptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *StringOptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, s := range f.vals {
		binary.LittleEndian.PutUint32(bs, uint32(len(s)))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
		buf.WriteString(s)
	}

	return f.DoWrite(w, meta, buf.Bytes(), len(f.Defs), f.stats)
}

func (f *StringOptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	for j := 0; j < f.Values(); j++ {
		var x int32
		if err := binary.Read(rr, binary.LittleEndian, &x); err != nil {
			return err
		}
		s := make([]byte, x)
		if _, err := rr.Read(s); err != nil {
			return err
		}

		f.vals = append(f.vals, string(s))
	}
	return nil
}

func (f *StringOptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type Float32Field struct {
	vals []float32
	parquet.RequiredField
	read  func(r Person) float32
	write func(r *Person, vals []float32)
	stats *float32stats
}

func NewFloat32Field(read func(r Person) float32, write func(r *Person, vals []float32), path []string, opts ...func(*parquet.RequiredField)) *Float32Field {
	return &Float32Field{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newFloat32stats(),
	}
}

func (f *Float32Field) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Float32Type, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *Float32Field) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]float32, int(pg.N))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Float32Field) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint32(bs, math.Float32bits(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *Float32Field) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *Float32Field) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *Float32Field) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type Float64Field struct {
	vals []float64
	parquet.RequiredField
	read  func(r Person) float64
	write func(r *Person, vals []float64)
	stats *float64stats
}

func NewFloat64Field(read func(r Person) float64, write func(r *Person, vals []float64), path []string, opts ...func(*parquet.RequiredField)) *Float64Field {
	return &Float64Field{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newFloat64stats(),
	}
}

func (f *Float64Field) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Float64Type, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *Float64Field) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]float64, int(pg.N))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Float64Field) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 8)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint64(bs, math.Float64bits(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *Float64Field) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *Float64Field) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *Float64Field) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type Float32OptionalField struct {
	parquet.OptionalField
	vals  []float32
	read  func(r Person, vals []float32, defs, reps []uint8) ([]float32, []uint8, []uint8)
	write func(r *Person, vals []float32, defs, reps []uint8) (int, int)
	stats *float32optionalStats
}

func NewFloat32OptionalField(read func(r Person, vals []float32, defs, reps []uint8) ([]float32, []uint8, []uint8), write func(r *Person, vals []float32, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *Float32OptionalField {
	return &Float32OptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newfloat32optionalStats(maxDef(types)),
	}
}

func (f *Float32OptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Float32Type, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *Float32OptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint32(bs, math.Float32bits(v))
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.Defs), f.stats)
}

func (f *Float32OptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]float32, f.Values()-len(f.vals))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Float32OptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *Float32OptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *Float32OptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type BoolOptionalField struct {
	parquet.OptionalField
	vals  []bool
	read  func(r Person, vals []bool, defs, reps []uint8) ([]bool, []uint8, []uint8)
	write func(r *Person, vals []bool, defs, reps []uint8) (int, int)
	stats *boolOptionalStats
}

func NewBoolOptionalField(read func(r Person, vals []bool, defs, reps []uint8) ([]bool, []uint8, []uint8), write func(r *Person, vals []bool, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *BoolOptionalField {
	return &BoolOptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newBoolOptionalStats(maxDef(types)),
	}
}

func (f *BoolOptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: BoolType, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *BoolOptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, sizes, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v, err := parquet.GetBools(rr, f.Values()-len(f.vals), sizes)
	f.vals = append(f.vals, v...)
	return err
}

func (f *BoolOptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *BoolOptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *BoolOptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	ln := len(f.vals)
	byteNum := (ln + 7) / 8
	rawBuf := make([]byte, byteNum)

	for i := 0; i < ln; i++ {
		if f.vals[i] {
			rawBuf[i/8] = rawBuf[i/8] | (1 << uint32(i%8))
		}
	}

	return f.DoWrite(w, meta, rawBuf, len(f.Defs), f.stats)
}

func (f *BoolOptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type Uint32Field struct {
	vals []uint32
	parquet.RequiredField
	read  func(r Person) uint32
	write func(r *Person, vals []uint32)
	stats *uint32stats
}

func NewUint32Field(read func(r Person) uint32, write func(r *Person, vals []uint32), path []string, opts ...func(*parquet.RequiredField)) *Uint32Field {
	return &Uint32Field{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
		stats:         newUint32stats(),
	}
}

func (f *Uint32Field) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Uint32Type, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *Uint32Field) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]uint32, int(pg.N))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Uint32Field) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 4)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint32(bs, v)
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.vals), f.stats)
}

func (f *Uint32Field) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *Uint32Field) Add(r Person) {
	v := f.read(r)
	f.stats.add(v)
	f.vals = append(f.vals, v)
}

func (f *Uint32Field) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type Uint64OptionalField struct {
	parquet.OptionalField
	vals  []uint64
	read  func(r Person, vals []uint64, defs, reps []uint8) ([]uint64, []uint8, []uint8)
	write func(r *Person, vals []uint64, defs, reps []uint8) (int, int)
	stats *uint64optionalStats
}

func NewUint64OptionalField(read func(r Person, vals []uint64, defs, reps []uint8) ([]uint64, []uint8, []uint8), write func(r *Person, vals []uint64, defs, reps []uint8) (int, int), path []string, types []int, opts ...func(*parquet.OptionalField)) *Uint64OptionalField {
	return &Uint64OptionalField{
		read:          read,
		write:         write,
		OptionalField: parquet.NewOptionalField(path, types, opts...),
		stats:         newuint64optionalStats(maxDef(types)),
	}
}

func (f *Uint64OptionalField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: Uint64Type, RepetitionType: f.RepetitionType, Types: f.Types}
}

func (f *Uint64OptionalField) Write(w io.Writer, meta *parquet.Metadata) error {
	buf := buffpool.Get()
	defer buffpool.Put(buf)

	bs := make([]byte, 8)
	for _, v := range f.vals {
		binary.LittleEndian.PutUint64(bs, v)
		if _, err := buf.Write(bs); err != nil {
			return err
		}
	}
	return f.DoWrite(w, meta, buf.Bytes(), len(f.Defs), f.stats)
}

func (f *Uint64OptionalField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, _, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	v := make([]uint64, f.Values()-len(f.vals))
	err = binary.Read(rr, binary.LittleEndian, &v)
	f.vals = append(f.vals, v...)
	return err
}

func (f *Uint64OptionalField) Add(r Person) {
	vals, defs, reps := f.read(r, f.vals, f.Defs, f.Reps)
	f.stats.add(vals[len(f.vals):], defs[len(f.Defs):])
	f.vals = vals
	f.Defs = defs
	f.Reps = reps
}

func (f *Uint64OptionalField) Scan(r *Person) {
	if len(f.Defs) == 0 {
		return
	}

	v, l := f.write(r, f.vals, f.Defs, f.Reps)
	f.vals = f.vals[v:]
	f.Defs = f.Defs[l:]
	if len(f.Reps) > 0 {
		f.Reps = f.Reps[l:]
	}
}

func (f *Uint64OptionalField) Levels() ([]uint8, []uint8) {
	return f.Defs, f.Reps
}

type BoolField struct {
	parquet.RequiredField
	vals  []bool
	read  func(r Person) bool
	write func(r *Person, vals []bool)
	stats *boolStats
}

func NewBoolField(read func(r Person) bool, write func(r *Person, vals []bool), path []string, opts ...func(*parquet.RequiredField)) *BoolField {
	return &BoolField{
		read:          read,
		write:         write,
		RequiredField: parquet.NewRequiredField(path, opts...),
	}
}

func (f *BoolField) Schema() parquet.Field {
	return parquet.Field{Name: f.Name(), Path: f.Path(), Type: BoolType, RepetitionType: parquet.RepetitionRequired, Types: []int{0}}
}

func (f *BoolField) Write(w io.Writer, meta *parquet.Metadata) error {
	ln := len(f.vals)
	n := (ln + 7) / 8
	rawBuf := make([]byte, n)

	for i := 0; i < ln; i++ {
		if f.vals[i] {
			rawBuf[i/8] = rawBuf[i/8] | (1 << uint32(i%8))
		}
	}

	return f.DoWrite(w, meta, rawBuf, len(f.vals), newBoolStats())
}

func (f *BoolField) Read(r io.ReadSeeker, pg parquet.Page) error {
	rr, sizes, err := f.DoRead(r, pg)
	if err != nil {
		return err
	}

	f.vals, err = parquet.GetBools(rr, int(pg.N), sizes)
	return err
}

func (f *BoolField) Scan(r *Person) {
	if len(f.vals) == 0 {
		return
	}

	f.write(r, f.vals)
	f.vals = f.vals[1:]
}

func (f *BoolField) Add(r Person) {
	v := f.read(r)
	f.vals = append(f.vals, v)
}

func (f *BoolField) Levels() ([]uint8, []uint8) {
	return nil, nil
}

type int32stats struct {
	min int32
	max int32
}

func newInt32stats() *int32stats {
	return &int32stats{
		min: int32(math.MaxInt32),
	}
}

func (i *int32stats) add(val int32) {
	if val < i.min {
		i.min = val
	}
	if val > i.max {
		i.max = val
	}
}

func (f *int32stats) bytes(v int32) []byte {
	bs := make([]byte, 4)
	binary.LittleEndian.PutUint32(bs, uint32(v))
	return bs
}

func (f *int32stats) NullCount() *int64 {
	return nil
}

func (f *int32stats) DistinctCount() *int64 {
	return nil
}

func (f *int32stats) Min() []byte {
	return f.bytes(f.min)
}

func (f *int32stats) Max() []byte {
	return f.bytes(f.max)
}

const nilString = "__#NIL#__"

type stringStats struct {
	min string
	max string
}

func newStringStats() *stringStats {
	return &stringStats{
		min: nilString,
		max: nilString,
	}
}

func (s *stringStats) add(val string) {
	if s.min == nilString {
		s.min = val
	} else {
		if val < s.min {
			s.min = val
		}
	}
	if s.max == nilString {
		s.max = val
	} else {
		if val > s.max {
			s.max = val
		}
	}
}

func (s *stringStats) NullCount() *int64 {
	return nil
}

func (s *stringStats) DistinctCount() *int64 {
	return nil
}

func (s *stringStats) Min() []byte {
	if s.min == nilString {
		return nil
	}
	return []byte(s.min)
}

func (s *stringStats) Max() []byte {
	if s.max == nilString {
		return nil
	}
	return []byte(s.max)
}

type int32optionalStats struct {
	min     int32
	max     int32
	nils    int64
	nonNils int64
	maxDef  uint8
}

func newint32optionalStats(d uint8) *int32optionalStats {
	return &int32optionalStats{
		min:    int32(math.MaxInt32),
		maxDef: d,
	}
}

func (f *int32optionalStats) add(vals []int32, defs []uint8) {
	var i int
	for _, def := range defs {
		if def < f.maxDef {
			f.nils++
		} else {
			val := vals[i]
			i++

			f.nonNils++
			if val < f.min {
				f.min = val
			}
			if val > f.max {
				f.max = val
			}
		}
	}
}

func (f *int32optionalStats) bytes(v int32) []byte {
	bs := make([]byte, 4)
	binary.LittleEndian.PutUint32(bs, uint32(v))
	return bs
}

func (f *int32optionalStats) NullCount() *int64 {
	return &f.nils
}

func (f *int32optionalStats) DistinctCount() *int64 {
	return nil
}

func (f *int32optionalStats) Min() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.min)
}

func (f *int32optionalStats) Max() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.max)
}

type int64stats struct {
	min int64
	max int64
}

func newInt64stats() *int64stats {
	return &int64stats{
		min: int64(math.MaxInt64),
	}
}

func (i *int64stats) add(val int64) {
	if val < i.min {
		i.min = val
	}
	if val > i.max {
		i.max = val
	}
}

func (f *int64stats) bytes(v int64) []byte {
	bs := make([]byte, 8)
	binary.LittleEndian.PutUint64(bs, uint64(v))
	return bs
}

func (f *int64stats) NullCount() *int64 {
	return nil
}

func (f *int64stats) DistinctCount() *int64 {
	return nil
}

func (f *int64stats) Min() []byte {
	return f.bytes(f.min)
}

func (f *int64stats) Max() []byte {
	return f.bytes(f.max)
}

type int64optionalStats struct {
	min     int64
	max     int64
	nils    int64
	nonNils int64
	maxDef  uint8
}

func newint64optionalStats(d uint8) *int64optionalStats {
	return &int64optionalStats{
		min:    int64(math.MaxInt64),
		maxDef: d,
	}
}

func (f *int64optionalStats) add(vals []int64, defs []uint8) {
	var i int
	for _, def := range defs {
		if def < f.maxDef {
			f.nils++
		} else {
			val := vals[i]
			i++

			f.nonNils++
			if val < f.min {
				f.min = val
			}
			if val > f.max {
				f.max = val
			}
		}
	}
}

func (f *int64optionalStats) bytes(v int64) []byte {
	bs := make([]byte, 8)
	binary.LittleEndian.PutUint64(bs, uint64(v))
	return bs
}

func (f *int64optionalStats) NullCount() *int64 {
	return &f.nils
}

func (f *int64optionalStats) DistinctCount() *int64 {
	return nil
}

func (f *int64optionalStats) Min() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.min)
}

func (f *int64optionalStats) Max() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.max)
}

const nilOptString = "__#NIL#__"

type stringOptionalStats struct {
	min    string
	max    string
	nils   int64
	maxDef uint8
}

func newStringOptionalStats(d uint8) *stringOptionalStats {
	return &stringOptionalStats{
		min:    nilOptString,
		max:    nilOptString,
		maxDef: d,
	}
}

func (s *stringOptionalStats) add(vals []string, defs []uint8) {
	var i int
	for _, def := range defs {
		if def < s.maxDef {
			s.nils++
		} else {
			val := vals[i]
			if s.min == nilOptString {
				s.min = val
			} else {
				if val < s.min {
					s.min = val
				}
			}
			if s.max == nilOptString {
				s.max = val
			} else {
				if val > s.max {
					s.max = val
				}
			}
			i++
		}
	}
}

func (s *stringOptionalStats) NullCount() *int64 {
	return &s.nils
}

func (s *stringOptionalStats) DistinctCount() *int64 {
	return nil
}

func (s *stringOptionalStats) Min() []byte {
	if s.min == nilOptString {
		return nil
	}
	return []byte(s.min)
}

func (s *stringOptionalStats) Max() []byte {
	if s.max == nilOptString {
		return nil
	}
	return []byte(s.max)
}

type float32stats struct {
	min float32
	max float32
}

func newFloat32stats() *float32stats {
	return &float32stats{
		min: float32(math.MaxFloat32),
	}
}

func (i *float32stats) add(val float32) {
	if val < i.min {
		i.min = val
	}
	if val > i.max {
		i.max = val
	}
}

func (f *float32stats) bytes(v float32) []byte {
	bs := make([]byte, 4)
	binary.LittleEndian.PutUint32(bs, math.Float32bits(v))
	return bs
}

func (f *float32stats) NullCount() *int64 {
	return nil
}

func (f *float32stats) DistinctCount() *int64 {
	return nil
}

func (f *float32stats) Min() []byte {
	return f.bytes(f.min)
}

func (f *float32stats) Max() []byte {
	return f.bytes(f.max)
}

type float64stats struct {
	min float64
	max float64
}

func newFloat64stats() *float64stats {
	return &float64stats{
		min: float64(math.MaxFloat64),
	}
}

func (i *float64stats) add(val float64) {
	if val < i.min {
		i.min = val
	}
	if val > i.max {
		i.max = val
	}
}

func (f *float64stats) bytes(v float64) []byte {
	bs := make([]byte, 8)
	binary.LittleEndian.PutUint64(bs, math.Float64bits(v))
	return bs
}

func (f *float64stats) NullCount() *int64 {
	return nil
}

func (f *float64stats) DistinctCount() *int64 {
	return nil
}

func (f *float64stats) Min() []byte {
	return f.bytes(f.min)
}

func (f *float64stats) Max() []byte {
	return f.bytes(f.max)
}

type float32optionalStats struct {
	min     float32
	max     float32
	nils    int64
	nonNils int64
	maxDef  uint8
}

func newfloat32optionalStats(d uint8) *float32optionalStats {
	return &float32optionalStats{
		min:    float32(math.MaxFloat32),
		maxDef: d,
	}
}

func (f *float32optionalStats) add(vals []float32, defs []uint8) {
	var i int
	for _, def := range defs {
		if def < f.maxDef {
			f.nils++
		} else {
			val := vals[i]
			i++

			f.nonNils++
			if val < f.min {
				f.min = val
			}
			if val > f.max {
				f.max = val
			}
		}
	}
}

func (f *float32optionalStats) bytes(v float32) []byte {
	bs := make([]byte, 4)
	binary.LittleEndian.PutUint32(bs, math.Float32bits(v))
	return bs
}

func (f *float32optionalStats) NullCount() *int64 {
	return &f.nils
}

func (f *float32optionalStats) DistinctCount() *int64 {
	return nil
}

func (f *float32optionalStats) Min() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.min)
}

func (f *float32optionalStats) Max() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.max)
}

type boolOptionalStats struct {
	maxDef uint8
	nils   int64
}

func newBoolOptionalStats(d uint8) *boolOptionalStats {
	return &boolOptionalStats{maxDef: d}
}

func (b *boolOptionalStats) add(vals []bool, defs []uint8) {
	for _, def := range defs {
		if def < b.maxDef {
			b.nils++
		}
	}
}

func (b *boolOptionalStats) NullCount() *int64 {
	return &b.nils
}

func (b *boolOptionalStats) DistinctCount() *int64 {
	return nil
}

func (b *boolOptionalStats) Min() []byte {
	return nil
}

func (b *boolOptionalStats) Max() []byte {
	return nil
}

type uint32stats struct {
	min uint32
	max uint32
}

func newUint32stats() *uint32stats {
	return &uint32stats{
		min: uint32(math.MaxUint32),
	}
}

func (i *uint32stats) add(val uint32) {
	if val < i.min {
		i.min = val
	}
	if val > i.max {
		i.max = val
	}
}

func (f *uint32stats) bytes(v uint32) []byte {
	bs := make([]byte, 4)
	binary.LittleEndian.PutUint32(bs, v)
	return bs
}

func (f *uint32stats) NullCount() *int64 {
	return nil
}

func (f *uint32stats) DistinctCount() *int64 {
	return nil
}

func (f *uint32stats) Min() []byte {
	return f.bytes(f.min)
}

func (f *uint32stats) Max() []byte {
	return f.bytes(f.max)
}

type uint64optionalStats struct {
	min     uint64
	max     uint64
	nils    int64
	nonNils int64
	maxDef  uint8
}

func newuint64optionalStats(d uint8) *uint64optionalStats {
	return &uint64optionalStats{
		min:    uint64(math.MaxUint64),
		maxDef: d,
	}
}

func (f *uint64optionalStats) add(vals []uint64, defs []uint8) {
	var i int
	for _, def := range defs {
		if def < f.maxDef {
			f.nils++
		} else {
			val := vals[i]
			i++

			f.nonNils++
			if val < f.min {
				f.min = val
			}
			if val > f.max {
				f.max = val
			}
		}
	}
}

func (f *uint64optionalStats) bytes(v uint64) []byte {
	bs := make([]byte, 8)
	binary.LittleEndian.PutUint64(bs, v)
	return bs
}

func (f *uint64optionalStats) NullCount() *int64 {
	return &f.nils
}

func (f *uint64optionalStats) DistinctCount() *int64 {
	return nil
}

func (f *uint64optionalStats) Min() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.min)
}

func (f *uint64optionalStats) Max() []byte {
	if f.nonNils == 0 {
		return nil
	}
	return f.bytes(f.max)
}

type boolStats struct{}

func newBoolStats() *boolStats             { return &boolStats{} }
func (b *boolStats) NullCount() *int64     { return nil }
func (b *boolStats) DistinctCount() *int64 { return nil }
func (b *boolStats) Min() []byte           { return nil }
func (b *boolStats) Max() []byte           { return nil }

func pint32(i int32) *int32       { return &i }
func puint32(i uint32) *uint32    { return &i }
func pint64(i int64) *int64       { return &i }
func puint64(i uint64) *uint64    { return &i }
func pbool(b bool) *bool          { return &b }
func pstring(s string) *string    { return &s }
func pfloat32(f float32) *float32 { return &f }
func pfloat64(f float64) *float64 { return &f }

// keeps track of the indices of repeated fields
// that have already been handled by a previous field
type indices []int

func (i indices) rep(rep uint8) {
	if rep > 0 {
		r := int(rep) - 1
		i[r] = i[r] + 1
		for j := int(rep); j < len(i); j++ {
			i[j] = 0
		}
	}
}

func maxDef(types []int) uint8 {
	var out uint8
	for _, typ := range types {
		if typ > 0 {
			out++
		}
	}
	return out
}

func Int32Type(se *sch.SchemaElement) {
	t := sch.Type_INT32
	se.Type = &t
}

func Uint32Type(se *sch.SchemaElement) {
	t := sch.Type_INT32
	se.Type = &t
	ct := sch.ConvertedType_UINT_32
	se.ConvertedType = &ct
}

func Int64Type(se *sch.SchemaElement) {
	t := sch.Type_INT64
	se.Type = &t
}

func Uint64Type(se *sch.SchemaElement) {
	t := sch.Type_INT64
	se.Type = &t
	ct := sch.ConvertedType_UINT_64
	se.ConvertedType = &ct
}

func Float32Type(se *sch.SchemaElement) {
	t := sch.Type_FLOAT
	se.Type = &t
}

func Float64Type(se *sch.SchemaElement) {
	t := sch.Type_DOUBLE
	se.Type = &t
}

func BoolType(se *sch.SchemaElement) {
	t := sch.Type_BOOLEAN
	se.Type = &t
}

func StringType(se *sch.SchemaElement) {
	t := sch.Type_BYTE_ARRAY
	se.Type = &t
}