parquet/metadata/statistics_types.gen.go

// Code generated by statistics_types.gen.go.tmpl. DO NOT EDIT.

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package metadata

import (
	"math"

	"github.com/apache/arrow/go/v6/arrow"
	"github.com/apache/arrow/go/v6/arrow/memory"
	"github.com/apache/arrow/go/v6/parquet"
	"github.com/apache/arrow/go/v6/parquet/internal/encoding"
	"github.com/apache/arrow/go/v6/parquet/internal/utils"
	"github.com/apache/arrow/go/v6/parquet/schema"
	"golang.org/x/xerrors"
)

type minmaxPairInt32 [2]int32

// Int32Statistics is the typed interface for managing stats for a column
// of Int32 type.
type Int32Statistics struct {
	statistics
	min int32
	max int32

	bitSetReader utils.SetBitRunReader
}

// NewInt32Statistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Int32
func NewInt32Statistics(descr *schema.Column, mem memory.Allocator) *Int32Statistics {
	if descr.PhysicalType() != parquet.Types.Int32 {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Int32 stat object", descr.PhysicalType()))
	}

	return &Int32Statistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewInt32StatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewInt32StatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *Int32Statistics {
	ret := NewInt32Statistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *Int32Statistics) plainEncode(src int32) []byte {
	s.encoder.(encoding.Int32Encoder).Put([]int32{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *Int32Statistics) plainDecode(src []byte) int32 {
	var buf [1]int32

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.Int32Decoder).Decode(buf[:])
	return buf[0]
}

func (s *Int32Statistics) minval(a, b int32) int32 {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *Int32Statistics) maxval(a, b int32) int32 {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *Int32Statistics) MinMaxEqual(rhs *Int32Statistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *Int32Statistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*Int32Statistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *Int32Statistics) getMinMax(values []int32) (min, max int32) {
	if s.order == schema.SortSIGNED {
		min, max = utils.GetMinMaxInt32(values)
	} else {
		umin, umax := utils.GetMinMaxUint32(arrow.Uint32Traits.CastFromBytes(arrow.Int32Traits.CastToBytes(values)))
		min, max = int32(umin), int32(umax)
	}
	return
}

func (s *Int32Statistics) getMinMaxSpaced(values []int32, validBits []byte, validBitsOffset int64) (min, max int32) {
	min = s.defaultMin()
	max = s.defaultMax()
	var fn func([]int32) (int32, int32)
	if s.order == schema.SortSIGNED {
		fn = utils.GetMinMaxInt32
	} else {
		fn = func(v []int32) (int32, int32) {
			umin, umax := utils.GetMinMaxUint32(arrow.Uint32Traits.CastFromBytes(arrow.Int32Traits.CastToBytes(values)))
			return int32(umin), int32(umax)
		}
	}

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		localMin, localMax := fn(values[int(run.Pos):int(run.Pos+run.Length)])
		if min > localMin {
			min = localMin
		}
		if max < localMax {
			max = localMax
		}
	}
	return
}

func (s *Int32Statistics) Min() int32 { return s.min }
func (s *Int32Statistics) Max() int32 { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *Int32Statistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*Int32Statistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *Int32Statistics) Update(values []int32, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *Int32Statistics) UpdateSpaced(values []int32, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *Int32Statistics) SetMinMax(argMin, argMax int32) {
	maybeMinMax := s.cleanStat([2]int32{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *Int32Statistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *Int32Statistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *Int32Statistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairInt64 [2]int64

// Int64Statistics is the typed interface for managing stats for a column
// of Int64 type.
type Int64Statistics struct {
	statistics
	min int64
	max int64

	bitSetReader utils.SetBitRunReader
}

// NewInt64Statistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Int64
func NewInt64Statistics(descr *schema.Column, mem memory.Allocator) *Int64Statistics {
	if descr.PhysicalType() != parquet.Types.Int64 {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Int64 stat object", descr.PhysicalType()))
	}

	return &Int64Statistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewInt64StatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewInt64StatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *Int64Statistics {
	ret := NewInt64Statistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *Int64Statistics) plainEncode(src int64) []byte {
	s.encoder.(encoding.Int64Encoder).Put([]int64{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *Int64Statistics) plainDecode(src []byte) int64 {
	var buf [1]int64

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.Int64Decoder).Decode(buf[:])
	return buf[0]
}

func (s *Int64Statistics) minval(a, b int64) int64 {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *Int64Statistics) maxval(a, b int64) int64 {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *Int64Statistics) MinMaxEqual(rhs *Int64Statistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *Int64Statistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*Int64Statistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *Int64Statistics) getMinMax(values []int64) (min, max int64) {
	if s.order == schema.SortSIGNED {
		min, max = utils.GetMinMaxInt64(values)
	} else {
		umin, umax := utils.GetMinMaxUint64(arrow.Uint64Traits.CastFromBytes(arrow.Int64Traits.CastToBytes(values)))
		min, max = int64(umin), int64(umax)
	}
	return
}

func (s *Int64Statistics) getMinMaxSpaced(values []int64, validBits []byte, validBitsOffset int64) (min, max int64) {
	min = s.defaultMin()
	max = s.defaultMax()
	var fn func([]int64) (int64, int64)
	if s.order == schema.SortSIGNED {
		fn = utils.GetMinMaxInt64
	} else {
		fn = func(v []int64) (int64, int64) {
			umin, umax := utils.GetMinMaxUint64(arrow.Uint64Traits.CastFromBytes(arrow.Int64Traits.CastToBytes(values)))
			return int64(umin), int64(umax)
		}
	}

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		localMin, localMax := fn(values[int(run.Pos):int(run.Pos+run.Length)])
		if min > localMin {
			min = localMin
		}
		if max < localMax {
			max = localMax
		}
	}
	return
}

func (s *Int64Statistics) Min() int64 { return s.min }
func (s *Int64Statistics) Max() int64 { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *Int64Statistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*Int64Statistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *Int64Statistics) Update(values []int64, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *Int64Statistics) UpdateSpaced(values []int64, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *Int64Statistics) SetMinMax(argMin, argMax int64) {
	maybeMinMax := s.cleanStat([2]int64{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *Int64Statistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *Int64Statistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *Int64Statistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairInt96 [2]parquet.Int96

// Int96Statistics is the typed interface for managing stats for a column
// of Int96 type.
type Int96Statistics struct {
	statistics
	min parquet.Int96
	max parquet.Int96

	bitSetReader utils.SetBitRunReader
}

// NewInt96Statistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Int96
func NewInt96Statistics(descr *schema.Column, mem memory.Allocator) *Int96Statistics {
	if descr.PhysicalType() != parquet.Types.Int96 {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Int96 stat object", descr.PhysicalType()))
	}

	return &Int96Statistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewInt96StatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewInt96StatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *Int96Statistics {
	ret := NewInt96Statistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *Int96Statistics) plainEncode(src parquet.Int96) []byte {
	s.encoder.(encoding.Int96Encoder).Put([]parquet.Int96{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *Int96Statistics) plainDecode(src []byte) parquet.Int96 {
	var buf [1]parquet.Int96

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.Int96Decoder).Decode(buf[:])
	return buf[0]
}

func (s *Int96Statistics) minval(a, b parquet.Int96) parquet.Int96 {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *Int96Statistics) maxval(a, b parquet.Int96) parquet.Int96 {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *Int96Statistics) MinMaxEqual(rhs *Int96Statistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *Int96Statistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*Int96Statistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *Int96Statistics) getMinMax(values []parquet.Int96) (min, max parquet.Int96) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, v)
		max = s.maxval(max, v)
	}
	return
}

func (s *Int96Statistics) getMinMaxSpaced(values []parquet.Int96, validBits []byte, validBitsOffset int64) (min, max parquet.Int96) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, v)
			max = s.maxval(max, v)
		}
	}
	return
}

func (s *Int96Statistics) Min() parquet.Int96 { return s.min }
func (s *Int96Statistics) Max() parquet.Int96 { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *Int96Statistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*Int96Statistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *Int96Statistics) Update(values []parquet.Int96, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *Int96Statistics) UpdateSpaced(values []parquet.Int96, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *Int96Statistics) SetMinMax(argMin, argMax parquet.Int96) {
	maybeMinMax := s.cleanStat([2]parquet.Int96{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *Int96Statistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *Int96Statistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *Int96Statistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairFloat32 [2]float32

// Float32Statistics is the typed interface for managing stats for a column
// of Float32 type.
type Float32Statistics struct {
	statistics
	min float32
	max float32

	bitSetReader utils.SetBitRunReader
}

// NewFloat32Statistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Float
func NewFloat32Statistics(descr *schema.Column, mem memory.Allocator) *Float32Statistics {
	if descr.PhysicalType() != parquet.Types.Float {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Float32 stat object", descr.PhysicalType()))
	}

	return &Float32Statistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewFloat32StatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewFloat32StatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *Float32Statistics {
	ret := NewFloat32Statistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *Float32Statistics) plainEncode(src float32) []byte {
	s.encoder.(encoding.Float32Encoder).Put([]float32{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *Float32Statistics) plainDecode(src []byte) float32 {
	var buf [1]float32

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.Float32Decoder).Decode(buf[:])
	return buf[0]
}

func (s *Float32Statistics) minval(a, b float32) float32 {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *Float32Statistics) maxval(a, b float32) float32 {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *Float32Statistics) MinMaxEqual(rhs *Float32Statistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *Float32Statistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*Float32Statistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *Float32Statistics) coalesce(val, fallback float32) float32 {
	if math.IsNaN(float64(val)) {
		return fallback
	}
	return val
}

func (s *Float32Statistics) getMinMax(values []float32) (min, max float32) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, s.coalesce(v, defMin))
		max = s.maxval(max, s.coalesce(v, defMax))
	}
	return
}

func (s *Float32Statistics) getMinMaxSpaced(values []float32, validBits []byte, validBitsOffset int64) (min, max float32) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, coalesce(v, s.defaultMin()).(float32))
			max = s.maxval(max, coalesce(v, s.defaultMax()).(float32))
		}
	}
	return
}

func (s *Float32Statistics) Min() float32 { return s.min }
func (s *Float32Statistics) Max() float32 { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *Float32Statistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*Float32Statistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *Float32Statistics) Update(values []float32, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *Float32Statistics) UpdateSpaced(values []float32, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *Float32Statistics) SetMinMax(argMin, argMax float32) {
	maybeMinMax := s.cleanStat([2]float32{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *Float32Statistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *Float32Statistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *Float32Statistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairFloat64 [2]float64

// Float64Statistics is the typed interface for managing stats for a column
// of Float64 type.
type Float64Statistics struct {
	statistics
	min float64
	max float64

	bitSetReader utils.SetBitRunReader
}

// NewFloat64Statistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Double
func NewFloat64Statistics(descr *schema.Column, mem memory.Allocator) *Float64Statistics {
	if descr.PhysicalType() != parquet.Types.Double {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Float64 stat object", descr.PhysicalType()))
	}

	return &Float64Statistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewFloat64StatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewFloat64StatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *Float64Statistics {
	ret := NewFloat64Statistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *Float64Statistics) plainEncode(src float64) []byte {
	s.encoder.(encoding.Float64Encoder).Put([]float64{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *Float64Statistics) plainDecode(src []byte) float64 {
	var buf [1]float64

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.Float64Decoder).Decode(buf[:])
	return buf[0]
}

func (s *Float64Statistics) minval(a, b float64) float64 {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *Float64Statistics) maxval(a, b float64) float64 {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *Float64Statistics) MinMaxEqual(rhs *Float64Statistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *Float64Statistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*Float64Statistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *Float64Statistics) coalesce(val, fallback float64) float64 {
	if math.IsNaN(float64(val)) {
		return fallback
	}
	return val
}

func (s *Float64Statistics) getMinMax(values []float64) (min, max float64) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, s.coalesce(v, defMin))
		max = s.maxval(max, s.coalesce(v, defMax))
	}
	return
}

func (s *Float64Statistics) getMinMaxSpaced(values []float64, validBits []byte, validBitsOffset int64) (min, max float64) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, coalesce(v, s.defaultMin()).(float64))
			max = s.maxval(max, coalesce(v, s.defaultMax()).(float64))
		}
	}
	return
}

func (s *Float64Statistics) Min() float64 { return s.min }
func (s *Float64Statistics) Max() float64 { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *Float64Statistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*Float64Statistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *Float64Statistics) Update(values []float64, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *Float64Statistics) UpdateSpaced(values []float64, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *Float64Statistics) SetMinMax(argMin, argMax float64) {
	maybeMinMax := s.cleanStat([2]float64{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *Float64Statistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *Float64Statistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *Float64Statistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairBoolean [2]bool

// BooleanStatistics is the typed interface for managing stats for a column
// of Boolean type.
type BooleanStatistics struct {
	statistics
	min bool
	max bool

	bitSetReader utils.SetBitRunReader
}

// NewBooleanStatistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.Boolean
func NewBooleanStatistics(descr *schema.Column, mem memory.Allocator) *BooleanStatistics {
	if descr.PhysicalType() != parquet.Types.Boolean {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a Boolean stat object", descr.PhysicalType()))
	}

	return &BooleanStatistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewBooleanStatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewBooleanStatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *BooleanStatistics {
	ret := NewBooleanStatistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *BooleanStatistics) plainEncode(src bool) []byte {
	s.encoder.(encoding.BooleanEncoder).Put([]bool{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *BooleanStatistics) plainDecode(src []byte) bool {
	var buf [1]bool

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.BooleanDecoder).Decode(buf[:])
	return buf[0]
}

func (s *BooleanStatistics) minval(a, b bool) bool {
	if s.less(a, b) {
		return a
	}
	return b
}

func (s *BooleanStatistics) maxval(a, b bool) bool {
	if s.less(a, b) {
		return b
	}
	return a
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *BooleanStatistics) MinMaxEqual(rhs *BooleanStatistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *BooleanStatistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*BooleanStatistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *BooleanStatistics) getMinMax(values []bool) (min, max bool) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, v)
		max = s.maxval(max, v)
	}
	return
}

func (s *BooleanStatistics) getMinMaxSpaced(values []bool, validBits []byte, validBitsOffset int64) (min, max bool) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, v)
			max = s.maxval(max, v)
		}
	}
	return
}

func (s *BooleanStatistics) Min() bool { return s.min }
func (s *BooleanStatistics) Max() bool { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *BooleanStatistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*BooleanStatistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *BooleanStatistics) Update(values []bool, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *BooleanStatistics) UpdateSpaced(values []bool, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *BooleanStatistics) SetMinMax(argMin, argMax bool) {
	maybeMinMax := s.cleanStat([2]bool{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *BooleanStatistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *BooleanStatistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *BooleanStatistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairByteArray [2]parquet.ByteArray

// ByteArrayStatistics is the typed interface for managing stats for a column
// of ByteArray type.
type ByteArrayStatistics struct {
	statistics
	min parquet.ByteArray
	max parquet.ByteArray

	bitSetReader utils.SetBitRunReader
}

// NewByteArrayStatistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.ByteArray
func NewByteArrayStatistics(descr *schema.Column, mem memory.Allocator) *ByteArrayStatistics {
	if descr.PhysicalType() != parquet.Types.ByteArray {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a ByteArray stat object", descr.PhysicalType()))
	}

	return &ByteArrayStatistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},

		min: make([]byte, 0),
		max: make([]byte, 0),
	}
}

// NewByteArrayStatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewByteArrayStatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *ByteArrayStatistics {
	ret := NewByteArrayStatistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *ByteArrayStatistics) plainEncode(src parquet.ByteArray) []byte {
	return src
}

func (s *ByteArrayStatistics) plainDecode(src []byte) parquet.ByteArray {
	return src
}

func (s *ByteArrayStatistics) minval(a, b parquet.ByteArray) parquet.ByteArray {
	switch {
	case a == nil:
		return b
	case b == nil:
		return a
	case s.less(a, b):
		return a
	default:
		return b
	}
}

func (s *ByteArrayStatistics) maxval(a, b parquet.ByteArray) parquet.ByteArray {
	switch {
	case a == nil:
		return b
	case b == nil:
		return a
	case s.less(a, b):
		return b
	default:
		return a
	}
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *ByteArrayStatistics) MinMaxEqual(rhs *ByteArrayStatistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *ByteArrayStatistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*ByteArrayStatistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *ByteArrayStatistics) getMinMax(values []parquet.ByteArray) (min, max parquet.ByteArray) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, v)
		max = s.maxval(max, v)
	}
	return
}

func (s *ByteArrayStatistics) getMinMaxSpaced(values []parquet.ByteArray, validBits []byte, validBitsOffset int64) (min, max parquet.ByteArray) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, v)
			max = s.maxval(max, v)
		}
	}
	return
}

func (s *ByteArrayStatistics) Min() parquet.ByteArray { return s.min }
func (s *ByteArrayStatistics) Max() parquet.ByteArray { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *ByteArrayStatistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*ByteArrayStatistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *ByteArrayStatistics) Update(values []parquet.ByteArray, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *ByteArrayStatistics) UpdateSpaced(values []parquet.ByteArray, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *ByteArrayStatistics) SetMinMax(argMin, argMax parquet.ByteArray) {
	maybeMinMax := s.cleanStat([2]parquet.ByteArray{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *ByteArrayStatistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *ByteArrayStatistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *ByteArrayStatistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

type minmaxPairFixedLenByteArray [2]parquet.FixedLenByteArray

// FixedLenByteArrayStatistics is the typed interface for managing stats for a column
// of FixedLenByteArray type.
type FixedLenByteArrayStatistics struct {
	statistics
	min parquet.FixedLenByteArray
	max parquet.FixedLenByteArray

	bitSetReader utils.SetBitRunReader
}

// NewFixedLenByteArrayStatistics constructs an appropriate stat object type using the
// given column descriptor and allocator.
//
// Panics if the physical type of descr is not parquet.Type.FixedLenByteArray
func NewFixedLenByteArrayStatistics(descr *schema.Column, mem memory.Allocator) *FixedLenByteArrayStatistics {
	if descr.PhysicalType() != parquet.Types.FixedLenByteArray {
		panic(xerrors.Errorf("parquet: invalid type %s for constructing a FixedLenByteArray stat object", descr.PhysicalType()))
	}

	return &FixedLenByteArrayStatistics{
		statistics: statistics{
			descr:            descr,
			hasNullCount:     true,
			hasDistinctCount: true,
			order:            descr.SortOrder(),
			encoder:          encoding.NewEncoder(descr.PhysicalType(), parquet.Encodings.Plain, false, descr, mem),
			mem:              mem,
		},
	}
}

// NewFixedLenByteArrayStatisticsFromEncoded will construct a propertly typed statistics object
// initializing it with the provided information.
func NewFixedLenByteArrayStatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) *FixedLenByteArrayStatistics {
	ret := NewFixedLenByteArrayStatistics(descr, mem)
	ret.nvalues += nvalues
	if encoded.IsSetNullCount() {
		ret.incNulls(encoded.GetNullCount())
	}
	if encoded.IsSetDistinctCount() {
		ret.incDistinct(encoded.GetDistinctCount())
	}

	encodedMin := encoded.GetMin()
	if encodedMin != nil && len(encodedMin) > 0 {
		ret.min = ret.plainDecode(encodedMin)
	}
	encodedMax := encoded.GetMax()
	if encodedMax != nil && len(encodedMax) > 0 {
		ret.max = ret.plainDecode(encodedMax)
	}
	ret.hasMinMax = encoded.IsSetMax() || encoded.IsSetMin()
	return ret
}

func (s *FixedLenByteArrayStatistics) plainEncode(src parquet.FixedLenByteArray) []byte {
	s.encoder.(encoding.FixedLenByteArrayEncoder).Put([]parquet.FixedLenByteArray{src})
	buf, err := s.encoder.FlushValues()
	if err != nil {
		panic(err) // recovered by Encode
	}
	defer buf.Release()

	out := make([]byte, buf.Len())
	copy(out, buf.Bytes())
	return out
}

func (s *FixedLenByteArrayStatistics) plainDecode(src []byte) parquet.FixedLenByteArray {
	var buf [1]parquet.FixedLenByteArray

	decoder := encoding.NewDecoder(s.descr.PhysicalType(), parquet.Encodings.Plain, s.descr, s.mem)
	decoder.SetData(1, src)
	decoder.(encoding.FixedLenByteArrayDecoder).Decode(buf[:])
	return buf[0]
}

func (s *FixedLenByteArrayStatistics) minval(a, b parquet.FixedLenByteArray) parquet.FixedLenByteArray {
	switch {
	case a == nil:
		return b
	case b == nil:
		return a
	case s.less(a, b):
		return a
	default:
		return b
	}
}

func (s *FixedLenByteArrayStatistics) maxval(a, b parquet.FixedLenByteArray) parquet.FixedLenByteArray {
	switch {
	case a == nil:
		return b
	case b == nil:
		return a
	case s.less(a, b):
		return b
	default:
		return a
	}
}

// MinMaxEqual returns true if both stat objects have the same Min and Max values
func (s *FixedLenByteArrayStatistics) MinMaxEqual(rhs *FixedLenByteArrayStatistics) bool {
	return s.equal(s.min, rhs.min) && s.equal(s.max, rhs.max)
}

// Equals returns true only if both objects are the same type, have the same min and
// max values, null count, distinct count and number of values.
func (s *FixedLenByteArrayStatistics) Equals(other TypedStatistics) bool {
	if s.Type() != other.Type() {
		return false
	}
	rhs, ok := other.(*FixedLenByteArrayStatistics)
	if !ok {
		return false
	}

	if s.HasMinMax() != rhs.HasMinMax() {
		return false
	}
	return (s.hasMinMax && s.MinMaxEqual(rhs)) &&
		s.NullCount() == rhs.NullCount() &&
		s.DistinctCount() == rhs.DistinctCount() &&
		s.NumValues() == rhs.NumValues()
}

func (s *FixedLenByteArrayStatistics) getMinMax(values []parquet.FixedLenByteArray) (min, max parquet.FixedLenByteArray) {
	defMin := s.defaultMin()
	defMax := s.defaultMax()

	min = defMin
	max = defMax

	for _, v := range values {
		min = s.minval(min, v)
		max = s.maxval(max, v)
	}
	return
}

func (s *FixedLenByteArrayStatistics) getMinMaxSpaced(values []parquet.FixedLenByteArray, validBits []byte, validBitsOffset int64) (min, max parquet.FixedLenByteArray) {
	min = s.defaultMin()
	max = s.defaultMax()

	if s.bitSetReader == nil {
		s.bitSetReader = utils.NewSetBitRunReader(validBits, validBitsOffset, int64(len(values)))
	} else {
		s.bitSetReader.Reset(validBits, validBitsOffset, int64(len(values)))
	}

	for {
		run := s.bitSetReader.NextRun()
		if run.Length == 0 {
			break
		}
		for _, v := range values[int(run.Pos):int(run.Pos+run.Length)] {
			min = s.minval(min, v)
			max = s.maxval(max, v)
		}
	}
	return
}

func (s *FixedLenByteArrayStatistics) Min() parquet.FixedLenByteArray { return s.min }
func (s *FixedLenByteArrayStatistics) Max() parquet.FixedLenByteArray { return s.max }

// Merge merges the stats from other into this stat object, updating
// the null count, distinct count, number of values and the min/max if
// appropriate.
func (s *FixedLenByteArrayStatistics) Merge(other TypedStatistics) {
	rhs, ok := other.(*FixedLenByteArrayStatistics)
	if !ok {
		panic("incompatible stat type merge")
	}

	s.statistics.merge(rhs)
	if rhs.HasMinMax() {
		s.SetMinMax(rhs.Min(), rhs.Max())
	}
}

// Update is used to add more values to the current stat object, finding the
// min and max values etc.
func (s *FixedLenByteArrayStatistics) Update(values []parquet.FixedLenByteArray, numNull int64) {
	s.incNulls(numNull)
	s.nvalues += int64(len(values))

	if len(values) == 0 {
		return
	}

	s.SetMinMax(s.getMinMax(values))
}

// UpdateSpaced is just like Update, but for spaced values using validBits to determine
// and skip null values.
func (s *FixedLenByteArrayStatistics) UpdateSpaced(values []parquet.FixedLenByteArray, validBits []byte, validBitsOffset, numNull int64) {
	s.incNulls(numNull)
	notnull := int64(len(values)) - numNull
	s.nvalues += notnull

	if notnull == 0 {
		return
	}

	s.SetMinMax(s.getMinMaxSpaced(values, validBits, validBitsOffset))
}

// SetMinMax updates the min and max values only if they are not currently set
// or if argMin is less than the current min / argMax is greater than the current max
func (s *FixedLenByteArrayStatistics) SetMinMax(argMin, argMax parquet.FixedLenByteArray) {
	maybeMinMax := s.cleanStat([2]parquet.FixedLenByteArray{argMin, argMax})
	if maybeMinMax == nil {
		return
	}

	min := (*maybeMinMax)[0]
	max := (*maybeMinMax)[1]

	if !s.hasMinMax {
		s.hasMinMax = true
		s.min = min
		s.max = max
	} else {
		if !s.less(s.min, min) {
			s.min = min
		}
		if s.less(s.max, max) {
			s.max = max
		}
	}
}

// EncodeMin returns the encoded min value with plain encoding.
//
// ByteArray stats do not include the length in the encoding.
func (s *FixedLenByteArrayStatistics) EncodeMin() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.min)
	}
	return nil
}

// EncodeMax returns the current encoded max value with plain encoding
//
// ByteArray stats do not include the length in the encoding
func (s *FixedLenByteArrayStatistics) EncodeMax() []byte {
	if s.HasMinMax() {
		return s.plainEncode(s.max)
	}
	return nil
}

// Encode returns a populated EncodedStatistics object
func (s *FixedLenByteArrayStatistics) Encode() (enc EncodedStatistics, err error) {
	defer func() {
		if r := recover(); r != nil {
			switch r := r.(type) {
			case error:
				err = r
			case string:
				err = xerrors.New(r)
			default:
				err = xerrors.Errorf("unknown error type thrown from panic: %v", r)
			}
		}
	}()
	if s.HasMinMax() {
		enc.SetMax(s.EncodeMax())
		enc.SetMin(s.EncodeMin())
	}
	if s.HasNullCount() {
		enc.SetNullCount(s.NullCount())
	}
	if s.HasDistinctCount() {
		enc.SetDistinctCount(s.DistinctCount())
	}
	return
}

// NewStatistics uses the type in the column descriptor to construct the appropriate
// typed stats object. If mem is nil, then memory.DefaultAllocator will be used.
func NewStatistics(descr *schema.Column, mem memory.Allocator) TypedStatistics {
	if mem == nil {
		mem = memory.DefaultAllocator
	}
	switch descr.PhysicalType() {
	case parquet.Types.Int32:
		return NewInt32Statistics(descr, mem)
	case parquet.Types.Int64:
		return NewInt64Statistics(descr, mem)
	case parquet.Types.Int96:
		return NewInt96Statistics(descr, mem)
	case parquet.Types.Float:
		return NewFloat32Statistics(descr, mem)
	case parquet.Types.Double:
		return NewFloat64Statistics(descr, mem)
	case parquet.Types.Boolean:
		return NewBooleanStatistics(descr, mem)
	case parquet.Types.ByteArray:
		return NewByteArrayStatistics(descr, mem)
	case parquet.Types.FixedLenByteArray:
		return NewFixedLenByteArrayStatistics(descr, mem)
	default:
		panic("not implemented")
	}
}

// NewStatisticsFromEncoded uses the provided information to initialize a typed stat object
// by checking the type of the provided column descriptor.
//
// If mem is nil, then memory.DefaultAllocator is used.
func NewStatisticsFromEncoded(descr *schema.Column, mem memory.Allocator, nvalues int64, encoded StatProvider) TypedStatistics {
	if mem == nil {
		mem = memory.DefaultAllocator
	}
	switch descr.PhysicalType() {
	case parquet.Types.Int32:
		return NewInt32StatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.Int64:
		return NewInt64StatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.Int96:
		return NewInt96StatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.Float:
		return NewFloat32StatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.Double:
		return NewFloat64StatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.Boolean:
		return NewBooleanStatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.ByteArray:
		return NewByteArrayStatisticsFromEncoded(descr, mem, nvalues, encoded)
	case parquet.Types.FixedLenByteArray:
		return NewFixedLenByteArrayStatisticsFromEncoded(descr, mem, nvalues, encoded)
	default:
		panic("not implemented")
	}
}