// Code generated by typed_encoder.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 encoding

import (
	"errors"
	"fmt"
	"unsafe"

	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
	"github.com/apache/arrow-go/v18/internal/bitutils"
	shared_utils "github.com/apache/arrow-go/v18/internal/utils"
	"github.com/apache/arrow-go/v18/parquet"
	format "github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet"
	"github.com/apache/arrow-go/v18/parquet/internal/utils"
	"github.com/apache/arrow-go/v18/parquet/schema"
	"golang.org/x/xerrors"
)

// fully typed encoder interfaces to enable writing against encoder/decoders
// without having to care about what encoding type is actually being used.

var (
	Int32EncoderTraits             int32EncoderTraits
	Int32DecoderTraits             int32DecoderTraits
	Int64EncoderTraits             int64EncoderTraits
	Int64DecoderTraits             int64DecoderTraits
	Int96EncoderTraits             int96EncoderTraits
	Int96DecoderTraits             int96DecoderTraits
	Float32EncoderTraits           float32EncoderTraits
	Float32DecoderTraits           float32DecoderTraits
	Float64EncoderTraits           float64EncoderTraits
	Float64DecoderTraits           float64DecoderTraits
	BooleanEncoderTraits           boolEncoderTraits
	BooleanDecoderTraits           boolDecoderTraits
	ByteArrayEncoderTraits         byteArrayEncoderTraits
	ByteArrayDecoderTraits         byteArrayDecoderTraits
	FixedLenByteArrayEncoderTraits fixedLenByteArrayEncoderTraits
	FixedLenByteArrayDecoderTraits fixedLenByteArrayDecoderTraits
)

// Int32Encoder is the interface for all encoding types that implement encoding
// int32 values.
type Int32Encoder interface {
	TypedEncoder
	Put([]int32)
	PutSpaced([]int32, []byte, int64)
}

// Int32Decoder is the interface for all encoding types that implement decoding
// int32 values.
type Int32Decoder interface {
	TypedDecoder
	Decode([]int32) (int, error)
	DecodeSpaced([]int32, int, []byte, int64) (int, error)
}

// the int32EncoderTraits struct is used to make it easy to create encoders and decoders based on type
type int32EncoderTraits struct{}

// Encoder returns an encoder for int32 type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (int32EncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictInt32Encoder{newDictEncoderBase(descr, NewInt32Dictionary(), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainInt32Encoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_DELTA_BINARY_PACKED:
		return &DeltaBitPackInt32Encoder{
			encoder: newEncoderBase(e, descr, mem),
		}
	case format.Encoding_BYTE_STREAM_SPLIT:
		return &ByteStreamSplitInt32Encoder{PlainInt32Encoder: PlainInt32Encoder{encoder: newEncoderBase(e, descr, mem)}}
	default:
		panic("unimplemented encoding type")
	}
}

// int32DecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for int32 values
type int32DecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n int32 values.
func (int32DecoderTraits) BytesRequired(n int) int {
	return arrow.Int32Traits.BytesRequired(n)
}

// Decoder returns a decoder for int32 typed data of the requested encoding type if available
func (int32DecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictInt32Decoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainInt32Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.DeltaBinaryPacked:
		if mem == nil {
			mem = memory.DefaultAllocator
		}
		return &DeltaBitPackInt32Decoder{
			decoder: newDecoderBase(format.Encoding(e), descr),
			mem:     mem,
		}
	case parquet.Encodings.ByteStreamSplit:
		return &ByteStreamSplitInt32Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictInt32Encoder is an encoder for int32 data using dictionary encoding
type DictInt32Encoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictInt32Encoder) Type() parquet.Type {
	return parquet.Types.Int32
}

// WriteDict populates the byte slice with the dictionary index
func (enc *DictInt32Encoder) WriteDict(out []byte) {
	enc.memo.(NumericMemoTable).WriteOutLE(out)
}

// Put encodes the values passed in, adding to the index as needed.
func (enc *DictInt32Encoder) Put(in []int32) {
	for _, val := range in {
		enc.dictEncoder.Put(val)
	}
}

// PutSpaced is the same as Put but for when the data being encoded has slots open for
// null values, using the bitmap provided to skip values as needed.
func (enc *DictInt32Encoder) PutSpaced(in []int32, validBits []byte, validBitsOffset int64) {
	bitutils.VisitSetBitRuns(validBits, validBitsOffset, int64(len(in)), func(pos, length int64) error {
		for i := int64(0); i < length; i++ {
			enc.dictEncoder.Put(in[i+pos])
		}
		return nil
	})
}

// PutDictionary allows pre-seeding a dictionary encoder with
// a dictionary from an Arrow Array.
//
// The passed in array must not have any nulls and this can only
// be called on an empty encoder.
func (enc *DictInt32Encoder) PutDictionary(values arrow.Array) error {
	if err := enc.canPutDictionary(values); err != nil {
		return err
	}

	enc.dictEncodedSize += values.Len() * arrow.Int32SizeBytes
	data := values.(*array.Int32).Int32Values()
	for _, v := range data {
		if _, _, err := enc.memo.GetOrInsert(v); err != nil {
			return err
		}
	}

	values.Retain()
	enc.preservedDict = values
	return nil
}

// DictInt32Decoder is a decoder for decoding dictionary encoded data for int32 columns
type DictInt32Decoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictInt32Decoder) Type() parquet.Type {
	return parquet.Types.Int32
}

func (d *DictInt32Decoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictInt32Decoder) Decode(out []int32) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictInt32Decoder) DecodeSpaced(out []int32, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// Int32DictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type Int32DictConverter struct {
	valueDecoder Int32Decoder
	dict         []int32
	zeroVal      int32
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *Int32DictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]int32, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *Int32DictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *Int32DictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]int32)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for int32
func (dc *Int32DictConverter) FillZero(out interface{}) {
	o := out.([]int32)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *Int32DictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]int32)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// Int64Encoder is the interface for all encoding types that implement encoding
// int64 values.
type Int64Encoder interface {
	TypedEncoder
	Put([]int64)
	PutSpaced([]int64, []byte, int64)
}

// Int64Decoder is the interface for all encoding types that implement decoding
// int64 values.
type Int64Decoder interface {
	TypedDecoder
	Decode([]int64) (int, error)
	DecodeSpaced([]int64, int, []byte, int64) (int, error)
}

// the int64EncoderTraits struct is used to make it easy to create encoders and decoders based on type
type int64EncoderTraits struct{}

// Encoder returns an encoder for int64 type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (int64EncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictInt64Encoder{newDictEncoderBase(descr, NewInt64Dictionary(), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainInt64Encoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_DELTA_BINARY_PACKED:
		return &DeltaBitPackInt64Encoder{
			encoder: newEncoderBase(e, descr, mem),
		}
	case format.Encoding_BYTE_STREAM_SPLIT:
		return &ByteStreamSplitInt64Encoder{PlainInt64Encoder: PlainInt64Encoder{encoder: newEncoderBase(e, descr, mem)}}
	default:
		panic("unimplemented encoding type")
	}
}

// int64DecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for int64 values
type int64DecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n int64 values.
func (int64DecoderTraits) BytesRequired(n int) int {
	return arrow.Int64Traits.BytesRequired(n)
}

// Decoder returns a decoder for int64 typed data of the requested encoding type if available
func (int64DecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictInt64Decoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainInt64Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.DeltaBinaryPacked:
		if mem == nil {
			mem = memory.DefaultAllocator
		}
		return &DeltaBitPackInt64Decoder{
			decoder: newDecoderBase(format.Encoding(e), descr),
			mem:     mem,
		}
	case parquet.Encodings.ByteStreamSplit:
		return &ByteStreamSplitInt64Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictInt64Encoder is an encoder for int64 data using dictionary encoding
type DictInt64Encoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictInt64Encoder) Type() parquet.Type {
	return parquet.Types.Int64
}

// WriteDict populates the byte slice with the dictionary index
func (enc *DictInt64Encoder) WriteDict(out []byte) {
	enc.memo.(NumericMemoTable).WriteOutLE(out)
}

// Put encodes the values passed in, adding to the index as needed.
func (enc *DictInt64Encoder) Put(in []int64) {
	for _, val := range in {
		enc.dictEncoder.Put(val)
	}
}

// PutSpaced is the same as Put but for when the data being encoded has slots open for
// null values, using the bitmap provided to skip values as needed.
func (enc *DictInt64Encoder) PutSpaced(in []int64, validBits []byte, validBitsOffset int64) {
	bitutils.VisitSetBitRuns(validBits, validBitsOffset, int64(len(in)), func(pos, length int64) error {
		for i := int64(0); i < length; i++ {
			enc.dictEncoder.Put(in[i+pos])
		}
		return nil
	})
}

// PutDictionary allows pre-seeding a dictionary encoder with
// a dictionary from an Arrow Array.
//
// The passed in array must not have any nulls and this can only
// be called on an empty encoder.
func (enc *DictInt64Encoder) PutDictionary(values arrow.Array) error {
	if err := enc.canPutDictionary(values); err != nil {
		return err
	}

	enc.dictEncodedSize += values.Len() * arrow.Int64SizeBytes
	data := values.(*array.Int64).Int64Values()
	for _, v := range data {
		if _, _, err := enc.memo.GetOrInsert(v); err != nil {
			return err
		}
	}

	values.Retain()
	enc.preservedDict = values
	return nil
}

// DictInt64Decoder is a decoder for decoding dictionary encoded data for int64 columns
type DictInt64Decoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictInt64Decoder) Type() parquet.Type {
	return parquet.Types.Int64
}

func (d *DictInt64Decoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictInt64Decoder) Decode(out []int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictInt64Decoder) DecodeSpaced(out []int64, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// Int64DictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type Int64DictConverter struct {
	valueDecoder Int64Decoder
	dict         []int64
	zeroVal      int64
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *Int64DictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]int64, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *Int64DictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *Int64DictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]int64)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for int64
func (dc *Int64DictConverter) FillZero(out interface{}) {
	o := out.([]int64)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *Int64DictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]int64)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// Int96Encoder is the interface for all encoding types that implement encoding
// parquet.Int96 values.
type Int96Encoder interface {
	TypedEncoder
	Put([]parquet.Int96)
	PutSpaced([]parquet.Int96, []byte, int64)
}

// Int96Decoder is the interface for all encoding types that implement decoding
// parquet.Int96 values.
type Int96Decoder interface {
	TypedDecoder
	Decode([]parquet.Int96) (int, error)
	DecodeSpaced([]parquet.Int96, int, []byte, int64) (int, error)
}

// the int96EncoderTraits struct is used to make it easy to create encoders and decoders based on type
type int96EncoderTraits struct{}

// Encoder returns an encoder for int96 type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (int96EncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictInt96Encoder{newDictEncoderBase(descr, NewBinaryDictionary(mem), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainInt96Encoder{encoder: newEncoderBase(e, descr, mem)}
	default:
		panic("unimplemented encoding type")
	}
}

// int96DecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for int96 values
type int96DecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n int96 values.
func (int96DecoderTraits) BytesRequired(n int) int {
	return parquet.Int96Traits.BytesRequired(n)
}

// Decoder returns a decoder for int96 typed data of the requested encoding type if available
func (int96DecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictInt96Decoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainInt96Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictInt96Encoder is an encoder for parquet.Int96 data using dictionary encoding
type DictInt96Encoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictInt96Encoder) Type() parquet.Type {
	return parquet.Types.Int96
}

// WriteDict populates the byte slice with the dictionary index
func (enc *DictInt96Encoder) WriteDict(out []byte) {
	enc.memo.(BinaryMemoTable).CopyFixedWidthValues(0, parquet.Int96SizeBytes, out)
}

// Put encodes the values passed in, adding to the index as needed
func (enc *DictInt96Encoder) Put(in []parquet.Int96) {
	for _, v := range in {
		memoIdx, found, err := enc.memo.GetOrInsert(v)
		if err != nil {
			panic(err)
		}
		if !found {
			enc.dictEncodedSize += parquet.Int96SizeBytes
		}
		enc.addIndex(memoIdx)
	}
}

// PutSpaced is like Put but assumes space for nulls
func (enc *DictInt96Encoder) PutSpaced(in []parquet.Int96, validBits []byte, validBitsOffset int64) {
	bitutils.VisitSetBitRuns(validBits, validBitsOffset, int64(len(in)), func(pos, length int64) error {
		enc.Put(in[pos : pos+length])
		return nil
	})
}

// PutDictionary allows pre-seeding a dictionary encoder with
// a dictionary from an Arrow Array.
//
// The passed in array must not have any nulls and this can only
// be called on an empty encoder.
func (enc *DictInt96Encoder) PutDictionary(arrow.Array) error {
	return fmt.Errorf("%w: direct PutDictionary to Int96", arrow.ErrNotImplemented)
}

// DictInt96Decoder is a decoder for decoding dictionary encoded data for parquet.Int96 columns
type DictInt96Decoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictInt96Decoder) Type() parquet.Type {
	return parquet.Types.Int96
}

func (d *DictInt96Decoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictInt96Decoder) Decode(out []parquet.Int96) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictInt96Decoder) DecodeSpaced(out []parquet.Int96, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// Int96DictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type Int96DictConverter struct {
	valueDecoder Int96Decoder
	dict         []parquet.Int96
	zeroVal      parquet.Int96
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *Int96DictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]parquet.Int96, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *Int96DictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *Int96DictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]parquet.Int96)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for parquet.Int96
func (dc *Int96DictConverter) FillZero(out interface{}) {
	o := out.([]parquet.Int96)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *Int96DictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]parquet.Int96)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// Float32Encoder is the interface for all encoding types that implement encoding
// float32 values.
type Float32Encoder interface {
	TypedEncoder
	Put([]float32)
	PutSpaced([]float32, []byte, int64)
}

// Float32Decoder is the interface for all encoding types that implement decoding
// float32 values.
type Float32Decoder interface {
	TypedDecoder
	Decode([]float32) (int, error)
	DecodeSpaced([]float32, int, []byte, int64) (int, error)
}

// the float32EncoderTraits struct is used to make it easy to create encoders and decoders based on type
type float32EncoderTraits struct{}

// Encoder returns an encoder for float32 type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (float32EncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictFloat32Encoder{newDictEncoderBase(descr, NewFloat32Dictionary(), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainFloat32Encoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_BYTE_STREAM_SPLIT:
		return &ByteStreamSplitFloat32Encoder{PlainFloat32Encoder: PlainFloat32Encoder{encoder: newEncoderBase(e, descr, mem)}}
	default:
		panic("unimplemented encoding type")
	}
}

// float32DecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for float32 values
type float32DecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n float32 values.
func (float32DecoderTraits) BytesRequired(n int) int {
	return arrow.Float32Traits.BytesRequired(n)
}

// Decoder returns a decoder for float32 typed data of the requested encoding type if available
func (float32DecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictFloat32Decoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainFloat32Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.ByteStreamSplit:
		return &ByteStreamSplitFloat32Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictFloat32Encoder is an encoder for float32 data using dictionary encoding
type DictFloat32Encoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictFloat32Encoder) Type() parquet.Type {
	return parquet.Types.Float
}

// WriteDict populates the byte slice with the dictionary index
func (enc *DictFloat32Encoder) WriteDict(out []byte) {
	enc.memo.(NumericMemoTable).WriteOutLE(out)
}

// Put encodes the values passed in, adding to the index as needed.
func (enc *DictFloat32Encoder) Put(in []float32) {
	for _, val := range in {
		enc.dictEncoder.Put(val)
	}
}

// PutSpaced is the same as Put but for when the data being encoded has slots open for
// null values, using the bitmap provided to skip values as needed.
func (enc *DictFloat32Encoder) PutSpaced(in []float32, validBits []byte, validBitsOffset int64) {
	bitutils.VisitSetBitRuns(validBits, validBitsOffset, int64(len(in)), func(pos, length int64) error {
		for i := int64(0); i < length; i++ {
			enc.dictEncoder.Put(in[i+pos])
		}
		return nil
	})
}

// PutDictionary allows pre-seeding a dictionary encoder with
// a dictionary from an Arrow Array.
//
// The passed in array must not have any nulls and this can only
// be called on an empty encoder.
func (enc *DictFloat32Encoder) PutDictionary(values arrow.Array) error {
	if err := enc.canPutDictionary(values); err != nil {
		return err
	}

	enc.dictEncodedSize += values.Len() * arrow.Float32SizeBytes
	data := values.(*array.Float32).Float32Values()
	for _, v := range data {
		if _, _, err := enc.memo.GetOrInsert(v); err != nil {
			return err
		}
	}

	values.Retain()
	enc.preservedDict = values
	return nil
}

// DictFloat32Decoder is a decoder for decoding dictionary encoded data for float32 columns
type DictFloat32Decoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictFloat32Decoder) Type() parquet.Type {
	return parquet.Types.Float
}

func (d *DictFloat32Decoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictFloat32Decoder) Decode(out []float32) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictFloat32Decoder) DecodeSpaced(out []float32, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// Float32DictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type Float32DictConverter struct {
	valueDecoder Float32Decoder
	dict         []float32
	zeroVal      float32
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *Float32DictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]float32, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *Float32DictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *Float32DictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]float32)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for float32
func (dc *Float32DictConverter) FillZero(out interface{}) {
	o := out.([]float32)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *Float32DictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]float32)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// Float64Encoder is the interface for all encoding types that implement encoding
// float64 values.
type Float64Encoder interface {
	TypedEncoder
	Put([]float64)
	PutSpaced([]float64, []byte, int64)
}

// Float64Decoder is the interface for all encoding types that implement decoding
// float64 values.
type Float64Decoder interface {
	TypedDecoder
	Decode([]float64) (int, error)
	DecodeSpaced([]float64, int, []byte, int64) (int, error)
}

// the float64EncoderTraits struct is used to make it easy to create encoders and decoders based on type
type float64EncoderTraits struct{}

// Encoder returns an encoder for float64 type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (float64EncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictFloat64Encoder{newDictEncoderBase(descr, NewFloat64Dictionary(), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainFloat64Encoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_BYTE_STREAM_SPLIT:
		return &ByteStreamSplitFloat64Encoder{PlainFloat64Encoder: PlainFloat64Encoder{encoder: newEncoderBase(e, descr, mem)}}
	default:
		panic("unimplemented encoding type")
	}
}

// float64DecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for float64 values
type float64DecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n float64 values.
func (float64DecoderTraits) BytesRequired(n int) int {
	return arrow.Float64Traits.BytesRequired(n)
}

// Decoder returns a decoder for float64 typed data of the requested encoding type if available
func (float64DecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictFloat64Decoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainFloat64Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.ByteStreamSplit:
		return &ByteStreamSplitFloat64Decoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictFloat64Encoder is an encoder for float64 data using dictionary encoding
type DictFloat64Encoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictFloat64Encoder) Type() parquet.Type {
	return parquet.Types.Double
}

// WriteDict populates the byte slice with the dictionary index
func (enc *DictFloat64Encoder) WriteDict(out []byte) {
	enc.memo.(NumericMemoTable).WriteOutLE(out)
}

// Put encodes the values passed in, adding to the index as needed.
func (enc *DictFloat64Encoder) Put(in []float64) {
	for _, val := range in {
		enc.dictEncoder.Put(val)
	}
}

// PutSpaced is the same as Put but for when the data being encoded has slots open for
// null values, using the bitmap provided to skip values as needed.
func (enc *DictFloat64Encoder) PutSpaced(in []float64, validBits []byte, validBitsOffset int64) {
	bitutils.VisitSetBitRuns(validBits, validBitsOffset, int64(len(in)), func(pos, length int64) error {
		for i := int64(0); i < length; i++ {
			enc.dictEncoder.Put(in[i+pos])
		}
		return nil
	})
}

// PutDictionary allows pre-seeding a dictionary encoder with
// a dictionary from an Arrow Array.
//
// The passed in array must not have any nulls and this can only
// be called on an empty encoder.
func (enc *DictFloat64Encoder) PutDictionary(values arrow.Array) error {
	if err := enc.canPutDictionary(values); err != nil {
		return err
	}

	enc.dictEncodedSize += values.Len() * arrow.Float64SizeBytes
	data := values.(*array.Float64).Float64Values()
	for _, v := range data {
		if _, _, err := enc.memo.GetOrInsert(v); err != nil {
			return err
		}
	}

	values.Retain()
	enc.preservedDict = values
	return nil
}

// DictFloat64Decoder is a decoder for decoding dictionary encoded data for float64 columns
type DictFloat64Decoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictFloat64Decoder) Type() parquet.Type {
	return parquet.Types.Double
}

func (d *DictFloat64Decoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictFloat64Decoder) Decode(out []float64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictFloat64Decoder) DecodeSpaced(out []float64, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// Float64DictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type Float64DictConverter struct {
	valueDecoder Float64Decoder
	dict         []float64
	zeroVal      float64
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *Float64DictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]float64, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *Float64DictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *Float64DictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]float64)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for float64
func (dc *Float64DictConverter) FillZero(out interface{}) {
	o := out.([]float64)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *Float64DictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]float64)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// BooleanEncoder is the interface for all encoding types that implement encoding
// bool values.
type BooleanEncoder interface {
	TypedEncoder
	Put([]bool)
	PutSpaced([]bool, []byte, int64)
}

// BooleanDecoder is the interface for all encoding types that implement decoding
// bool values.
type BooleanDecoder interface {
	TypedDecoder
	Decode([]bool) (int, error)
	DecodeSpaced([]bool, int, []byte, int64) (int, error)
}

// the boolEncoderTraits struct is used to make it easy to create encoders and decoders based on type
type boolEncoderTraits struct{}

// Encoder returns an encoder for bool type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
// dictionary encoding does not exist for this type and Encoder will panic if useDict is true
func (boolEncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		panic("parquet: no bool dictionary encoding")
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainBooleanEncoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_RLE:
		return &RleBooleanEncoder{encoder: newEncoderBase(e, descr, mem)}
	default:
		panic("unimplemented encoding type")
	}
}

// boolDecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for bool values
type boolDecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n bool values.
func (boolDecoderTraits) BytesRequired(n int) int {
	return arrow.BooleanTraits.BytesRequired(n)
}

// Decoder returns a decoder for bool typed data of the requested encoding type if available
func (boolDecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		panic("dictionary decoding unimplemented for bool")
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainBooleanDecoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.RLE:
		return &RleBooleanDecoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// ByteArrayEncoder is the interface for all encoding types that implement encoding
// parquet.ByteArray values.
type ByteArrayEncoder interface {
	TypedEncoder
	Put([]parquet.ByteArray)
	PutSpaced([]parquet.ByteArray, []byte, int64)
}

// ByteArrayDecoder is the interface for all encoding types that implement decoding
// parquet.ByteArray values.
type ByteArrayDecoder interface {
	TypedDecoder
	Decode([]parquet.ByteArray) (int, error)
	DecodeSpaced([]parquet.ByteArray, int, []byte, int64) (int, error)
}

// the byteArrayEncoderTraits struct is used to make it easy to create encoders and decoders based on type
type byteArrayEncoderTraits struct{}

// Encoder returns an encoder for byteArray type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (byteArrayEncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictByteArrayEncoder{newDictEncoderBase(descr, NewBinaryDictionary(mem), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainByteArrayEncoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_DELTA_LENGTH_BYTE_ARRAY:
		return &DeltaLengthByteArrayEncoder{
			encoder: newEncoderBase(e, descr, mem),
			lengthEncoder: &DeltaBitPackInt32Encoder{
				encoder: newEncoderBase(e, descr, mem),
			},
		}
	case format.Encoding_DELTA_BYTE_ARRAY:
		return &DeltaByteArrayEncoder{
			encoder: newEncoderBase(e, descr, mem),
		}
	default:
		panic("unimplemented encoding type")
	}
}

// byteArrayDecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for byteArray values
type byteArrayDecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n byteArray values.
func (byteArrayDecoderTraits) BytesRequired(n int) int {
	return parquet.ByteArrayTraits.BytesRequired(n)
}

// Decoder returns a decoder for byteArray typed data of the requested encoding type if available
func (byteArrayDecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictByteArrayDecoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainByteArrayDecoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.DeltaLengthByteArray:
		if mem == nil {
			mem = memory.DefaultAllocator
		}
		return &DeltaLengthByteArrayDecoder{
			decoder: newDecoderBase(format.Encoding(e), descr),
			mem:     mem,
		}
	case parquet.Encodings.DeltaByteArray:
		if mem == nil {
			mem = memory.DefaultAllocator
		}
		return &DeltaByteArrayDecoder{
			DeltaLengthByteArrayDecoder: &DeltaLengthByteArrayDecoder{
				decoder: newDecoderBase(format.Encoding(e), descr),
				mem:     mem,
			}}
	default:
		panic("unimplemented encoding type")
	}
}

// DictByteArrayEncoder is an encoder for parquet.ByteArray data using dictionary encoding
type DictByteArrayEncoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictByteArrayEncoder) Type() parquet.Type {
	return parquet.Types.ByteArray
}

// DictByteArrayDecoder is a decoder for decoding dictionary encoded data for parquet.ByteArray columns
type DictByteArrayDecoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictByteArrayDecoder) Type() parquet.Type {
	return parquet.Types.ByteArray
}

func (d *DictByteArrayDecoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictByteArrayDecoder) Decode(out []parquet.ByteArray) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictByteArrayDecoder) DecodeSpaced(out []parquet.ByteArray, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// ByteArrayDictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type ByteArrayDictConverter struct {
	valueDecoder ByteArrayDecoder
	dict         []parquet.ByteArray
	zeroVal      parquet.ByteArray
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *ByteArrayDictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]parquet.ByteArray, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *ByteArrayDictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *ByteArrayDictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]parquet.ByteArray)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for parquet.ByteArray
func (dc *ByteArrayDictConverter) FillZero(out interface{}) {
	o := out.([]parquet.ByteArray)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *ByteArrayDictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]parquet.ByteArray)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// FixedLenByteArrayEncoder is the interface for all encoding types that implement encoding
// parquet.FixedLenByteArray values.
type FixedLenByteArrayEncoder interface {
	TypedEncoder
	Put([]parquet.FixedLenByteArray)
	PutSpaced([]parquet.FixedLenByteArray, []byte, int64)
}

// FixedLenByteArrayDecoder is the interface for all encoding types that implement decoding
// parquet.FixedLenByteArray values.
type FixedLenByteArrayDecoder interface {
	TypedDecoder
	Decode([]parquet.FixedLenByteArray) (int, error)
	DecodeSpaced([]parquet.FixedLenByteArray, int, []byte, int64) (int, error)
}

// the fixedLenByteArrayEncoderTraits struct is used to make it easy to create encoders and decoders based on type
type fixedLenByteArrayEncoderTraits struct{}

// Encoder returns an encoder for fixedLenByteArray type data, using the specified encoding type and whether or not
// it should be dictionary encoded.
func (fixedLenByteArrayEncoderTraits) Encoder(e format.Encoding, useDict bool, descr *schema.Column, mem memory.Allocator) TypedEncoder {
	if useDict {
		return &DictFixedLenByteArrayEncoder{newDictEncoderBase(descr, NewBinaryDictionary(mem), mem)}
	}

	switch e {
	case format.Encoding_PLAIN:
		return &PlainFixedLenByteArrayEncoder{encoder: newEncoderBase(e, descr, mem)}
	case format.Encoding_BYTE_STREAM_SPLIT:
		return &ByteStreamSplitFixedLenByteArrayEncoder{PlainFixedLenByteArrayEncoder: PlainFixedLenByteArrayEncoder{encoder: newEncoderBase(e, descr, mem)}}
	default:
		panic("unimplemented encoding type")
	}
}

// fixedLenByteArrayDecoderTraits is a helper struct for providing information regardless of the type
// and used as a generic way to create a Decoder or Dictionary Decoder for fixedLenByteArray values
type fixedLenByteArrayDecoderTraits struct{}

// BytesRequired returns the number of bytes required to store n fixedLenByteArray values.
func (fixedLenByteArrayDecoderTraits) BytesRequired(n int) int {
	return parquet.FixedLenByteArrayTraits.BytesRequired(n)
}

// Decoder returns a decoder for fixedLenByteArray typed data of the requested encoding type if available
func (fixedLenByteArrayDecoderTraits) Decoder(e parquet.Encoding, descr *schema.Column, useDict bool, mem memory.Allocator) TypedDecoder {
	if useDict {
		return &DictFixedLenByteArrayDecoder{dictDecoder{decoder: newDecoderBase(format.Encoding_RLE_DICTIONARY, descr), mem: mem}}
	}

	switch e {
	case parquet.Encodings.Plain:
		return &PlainFixedLenByteArrayDecoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	case parquet.Encodings.ByteStreamSplit:
		return &ByteStreamSplitFixedLenByteArrayDecoder{decoder: newDecoderBase(format.Encoding(e), descr)}
	default:
		panic("unimplemented encoding type")
	}
}

// DictFixedLenByteArrayEncoder is an encoder for parquet.FixedLenByteArray data using dictionary encoding
type DictFixedLenByteArrayEncoder struct {
	dictEncoder
}

// Type returns the underlying physical type that can be encoded with this encoder
func (enc *DictFixedLenByteArrayEncoder) Type() parquet.Type {
	return parquet.Types.FixedLenByteArray
}

// DictFixedLenByteArrayDecoder is a decoder for decoding dictionary encoded data for parquet.FixedLenByteArray columns
type DictFixedLenByteArrayDecoder struct {
	dictDecoder
}

// Type returns the underlying physical type that can be decoded with this decoder
func (DictFixedLenByteArrayDecoder) Type() parquet.Type {
	return parquet.Types.FixedLenByteArray
}

func (d *DictFixedLenByteArrayDecoder) Discard(n int) (int, error) {
	n = min(n, d.nvals)
	discarded, err := d.discard(n)
	if err != nil {
		return discarded, err
	}
	if n != discarded {
		return discarded, errors.New("parquet: dict eof exception")
	}
	return n, nil
}

// Decode populates the passed in slice with min(len(out), remaining values) values,
// decoding using the dictionary to get the actual values. Returns the number of values
// actually decoded and any error encountered.
func (d *DictFixedLenByteArrayDecoder) Decode(out []parquet.FixedLenByteArray) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decode(out[:vals])
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict eof exception")
	}
	return vals, nil
}

// Decode spaced is like Decode but will space out the data leaving slots for null values
// based on the provided bitmap.
func (d *DictFixedLenByteArrayDecoder) DecodeSpaced(out []parquet.FixedLenByteArray, nullCount int, validBits []byte, validBitsOffset int64) (int, error) {
	vals := shared_utils.Min(len(out), d.nvals)
	decoded, err := d.decodeSpaced(out[:vals], nullCount, validBits, validBitsOffset)
	if err != nil {
		return decoded, err
	}
	if vals != decoded {
		return decoded, xerrors.New("parquet: dict spaced eof exception")
	}
	return vals, nil
}

// FixedLenByteArrayDictConverter is a helper for dictionary handling which is used for converting
// run length encoded indexes into the actual values that are stored in the dictionary index page.
type FixedLenByteArrayDictConverter struct {
	valueDecoder FixedLenByteArrayDecoder
	dict         []parquet.FixedLenByteArray
	zeroVal      parquet.FixedLenByteArray
}

// ensure validates that we've decoded dictionary values up to the index
// provided so that we don't need to decode the entire dictionary at start.
func (dc *FixedLenByteArrayDictConverter) ensure(idx utils.IndexType) error {
	if len(dc.dict) <= int(idx) {
		if cap(dc.dict) <= int(idx) {
			val := make([]parquet.FixedLenByteArray, int(idx+1)-len(dc.dict))
			n, err := dc.valueDecoder.Decode(val)
			if err != nil {
				return err
			}
			dc.dict = append(dc.dict, val[:n]...)
		} else {
			cur := len(dc.dict)
			n, err := dc.valueDecoder.Decode(dc.dict[cur : idx+1])
			if err != nil {
				return err
			}
			dc.dict = dc.dict[:cur+n]
		}
	}
	return nil
}

// IsValid verifies that the set of indexes passed in are all valid indexes
// in the dictionary and if necessary decodes dictionary indexes up to the index
// requested.
func (dc *FixedLenByteArrayDictConverter) IsValid(idxes ...utils.IndexType) bool {
	min, max := shared_utils.GetMinMaxInt32(*(*[]int32)(unsafe.Pointer(&idxes)))
	dc.ensure(utils.IndexType(max))

	return min >= 0 && int(min) < len(dc.dict) && int(max) >= 0 && int(max) < len(dc.dict)
}

// Fill populates the slice passed in entirely with the value at dictionary index indicated by val
func (dc *FixedLenByteArrayDictConverter) Fill(out interface{}, val utils.IndexType) error {
	o := out.([]parquet.FixedLenByteArray)
	if err := dc.ensure(val); err != nil {
		return err
	}
	o[0] = dc.dict[val]
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
	return nil
}

// FillZero populates the entire slice of out with the zero value for parquet.FixedLenByteArray
func (dc *FixedLenByteArrayDictConverter) FillZero(out interface{}) {
	o := out.([]parquet.FixedLenByteArray)
	o[0] = dc.zeroVal
	for i := 1; i < len(o); i *= 2 {
		copy(o[i:], o[:i])
	}
}

// Copy populates the slice provided with the values in the dictionary at the indexes
// in the vals slice.
func (dc *FixedLenByteArrayDictConverter) Copy(out interface{}, vals []utils.IndexType) error {
	o := out.([]parquet.FixedLenByteArray)
	for idx, val := range vals {
		o[idx] = dc.dict[val]
	}
	return nil
}

// NewDictConverter creates a dict converter of the appropriate type, using the passed in
// decoder as the decoder to decode the dictionary index.
func NewDictConverter(dict TypedDecoder) utils.DictionaryConverter {
	switch dict.Type() {
	case parquet.Types.Int32:
		return &Int32DictConverter{valueDecoder: dict.(Int32Decoder), dict: make([]int32, 0, dict.ValuesLeft())}
	case parquet.Types.Int64:
		return &Int64DictConverter{valueDecoder: dict.(Int64Decoder), dict: make([]int64, 0, dict.ValuesLeft())}
	case parquet.Types.Int96:
		return &Int96DictConverter{valueDecoder: dict.(Int96Decoder), dict: make([]parquet.Int96, 0, dict.ValuesLeft())}
	case parquet.Types.Float:
		return &Float32DictConverter{valueDecoder: dict.(Float32Decoder), dict: make([]float32, 0, dict.ValuesLeft())}
	case parquet.Types.Double:
		return &Float64DictConverter{valueDecoder: dict.(Float64Decoder), dict: make([]float64, 0, dict.ValuesLeft())}
	case parquet.Types.ByteArray:
		return &ByteArrayDictConverter{valueDecoder: dict.(ByteArrayDecoder), dict: make([]parquet.ByteArray, 0, dict.ValuesLeft())}
	case parquet.Types.FixedLenByteArray:
		return &FixedLenByteArrayDictConverter{valueDecoder: dict.(FixedLenByteArrayDecoder), dict: make([]parquet.FixedLenByteArray, 0, dict.ValuesLeft())}
	default:
		return nil
	}
}

// helper function to get encoding traits object for the physical type indicated
func getEncodingTraits(t parquet.Type) EncoderTraits {
	switch t {
	case parquet.Types.Int32:
		return Int32EncoderTraits
	case parquet.Types.Int64:
		return Int64EncoderTraits
	case parquet.Types.Int96:
		return Int96EncoderTraits
	case parquet.Types.Float:
		return Float32EncoderTraits
	case parquet.Types.Double:
		return Float64EncoderTraits
	case parquet.Types.Boolean:
		return BooleanEncoderTraits
	case parquet.Types.ByteArray:
		return ByteArrayEncoderTraits
	case parquet.Types.FixedLenByteArray:
		return FixedLenByteArrayEncoderTraits
	default:
		return nil
	}
}

// helper function to get decoding traits object for the physical type indicated
func getDecodingTraits(t parquet.Type) DecoderTraits {
	switch t {
	case parquet.Types.Int32:
		return Int32DecoderTraits
	case parquet.Types.Int64:
		return Int64DecoderTraits
	case parquet.Types.Int96:
		return Int96DecoderTraits
	case parquet.Types.Float:
		return Float32DecoderTraits
	case parquet.Types.Double:
		return Float64DecoderTraits
	case parquet.Types.Boolean:
		return BooleanDecoderTraits
	case parquet.Types.ByteArray:
		return ByteArrayDecoderTraits
	case parquet.Types.FixedLenByteArray:
		return FixedLenByteArrayDecoderTraits
	default:
		return nil
	}
}