// NewStringTreeReader returns a StringTreeReader implementation along with any error that occurs.s
func NewStringTreeReader(present, data, length, dictionary io.Reader, encoding *proto.ColumnEncoding) (StringTreeReader, error) {
	switch kind := encoding.GetKind(); kind {
	case proto.ColumnEncoding_DIRECT, proto.ColumnEncoding_DIRECT_V2:
		return NewStringDirectTreeReader(present, data, length, kind)
	case proto.ColumnEncoding_DICTIONARY, proto.ColumnEncoding_DICTIONARY_V2:
		return NewStringDictionaryTreeReader(present, data, length, dictionary, encoding)
	}
	return nil, fmt.Errorf("unsupported column encoding: %s", encoding.GetKind())
}

// NewIntegerTreeReader returns a new IntegerReader or an error if one occurs.
func NewIntegerTreeReader(present, data io.Reader, encoding *proto.ColumnEncoding) (*IntegerTreeReader, error) {
	ireader, err := createIntegerReader(encoding.GetKind(), data, true, false)
	if err != nil {
		return nil, err
	}
	return &IntegerTreeReader{
		NewBaseTreeReader(present),
		ireader,
	}, nil
}

func NewBinaryTreeReader(present, data, length io.Reader, encoding *proto.ColumnEncoding) (*BinaryTreeReader, error) {
	lengthReader, err := createIntegerReader(encoding.GetKind(), length, false, false)
	if err != nil {
		return nil, err
	}
	return &BinaryTreeReader{
		BaseTreeReader: NewBaseTreeReader(present),
		length:         lengthReader,
		data:           data,
	}, nil
}

func NewListTreeReader(present, length io.Reader, value TreeReader, encoding *proto.ColumnEncoding) (*ListTreeReader, error) {
	lengthReader, err := createIntegerReader(encoding.GetKind(), length, false, false)
	if err != nil {
		return nil, err
	}
	return &ListTreeReader{
		BaseTreeReader: NewBaseTreeReader(present),
		length:         lengthReader,
		value:          value,
	}, nil
}

// NewDecimalTreeReader returns a new instances of a DecimalTreeReader or an error if one occurs.
func NewDecimalTreeReader(present, data, secondary io.Reader, encoding *proto.ColumnEncoding, precision, scale int) (*DecimalTreeReader, error) {
	ireader, err := createIntegerReader(encoding.GetKind(), secondary, true, false)
	if err != nil {
		return nil, err
	}
	return &DecimalTreeReader{
		BaseTreeReader: NewBaseTreeReader(present),
		data:           bufio.NewReader(data),
		secondary:      ireader,
		precision:      precision,
		scale:          scale,
	}, nil
}

// NewTimestampTreeReader returns a new TimestampTreeReader along with any error that occurs.
func NewTimestampTreeReader(present, data, secondary io.Reader, encoding *proto.ColumnEncoding) (*TimestampTreeReader, error) {
	dataReader, err := createIntegerReader(encoding.GetKind(), data, true, false)
	if err != nil {
		return nil, err
	}
	secondaryReader, err := createIntegerReader(encoding.GetKind(), secondary, true, false)
	if err != nil {
		return nil, err
	}
	return &TimestampTreeReader{
		BaseTreeReader: NewBaseTreeReader(present),
		data:           dataReader,
		secondary:      secondaryReader,
	}, nil
}

func (s *StringDictionaryTreeReader) readDictionaryLength(length io.Reader, encoding *proto.ColumnEncoding) error {
	lreader, err := createIntegerReader(encoding.GetKind(), length, false, false)
	if err != nil {
		return err
	}
	var offset int
	for lreader.Next() {
		l := int(lreader.Int())
		s.dictionaryLength = append(s.dictionaryLength, l)
		s.dictionaryOffsets = append(s.dictionaryOffsets, offset)
		offset += l
	}
	if err := lreader.Err(); err != nil && err != io.EOF {
		return err
	}
	return nil
}

func NewStringDictionaryTreeReader(present, data, length, dictionary io.Reader, encoding *proto.ColumnEncoding) (*StringDictionaryTreeReader, error) {
	ireader, err := createIntegerReader(encoding.GetKind(), data, false, false)
	if err != nil {
		return nil, err
	}
	r := &StringDictionaryTreeReader{
		BaseTreeReader: NewBaseTreeReader(present),
		reader:         ireader,
	}
	if dictionary != nil && encoding != nil {
		err := r.readDictionaryStream(dictionary)
		if err != nil {
			return nil, err
		}
		if length != nil {
			err = r.readDictionaryLength(length, encoding)
			if err != nil {
				return nil, err
			}
		}
	}
	return r, nil
}