package parquet

import (
	"bufio"
	"bytes"
	"cmp"
	"encoding/binary"
	"fmt"
	"hash/crc32"
	"io"
	"math"
	"math/bits"
	"os"
	"reflect"
	"slices"
	"strings"

	"github.com/parquet-go/parquet-go/compress"
	"github.com/parquet-go/parquet-go/encoding"
	"github.com/parquet-go/parquet-go/encoding/plain"
	"github.com/parquet-go/parquet-go/encoding/thrift"
	"github.com/parquet-go/parquet-go/format"
	"github.com/parquet-go/parquet-go/internal/memory"
	"github.com/parquet-go/parquet-go/sparse"
)

const (
	// The uncompressed page size is stored as int32 and must not be larger than the
	// maximum int32 value (see format.PageHeader).
	maxUncompressedPageSize = math.MaxInt32
)

// GenericWriter is similar to a Writer but uses a type parameter to define the
// Go type representing the schema of rows being written.
//
// Using this type over Writer has multiple advantages:
//
//   - By leveraging type information, the Go compiler can provide greater
//     guarantees that the code is correct. For example, the parquet.Writer.Write
//     method accepts an argument of type interface{}, which delays type checking
//     until runtime. The parquet.GenericWriter[T].Write method ensures at
//     compile time that the values it receives will be of type T, reducing the
//     risk of introducing errors.
//
//   - Since type information is known at compile time, the implementation of
//     parquet.GenericWriter[T] can make safe assumptions, removing the need for
//     runtime validation of how the parameters are passed to its methods.
//     Optimizations relying on type information are more effective, some of the
//     writer's state can be precomputed at initialization, which was not possible
//     with parquet.Writer.
//
//   - The parquet.GenericWriter[T].Write method uses a data-oriented design,
//     accepting an slice of T instead of a single value, creating more
//     opportunities to amortize the runtime cost of abstractions.
//     This optimization is not available for parquet.Writer because its Write
//     method's argument would be of type []interface{}, which would require
//     conversions back and forth from concrete types to empty interfaces (since
//     a []T cannot be interpreted as []interface{} in Go), would make the API
//     more difficult to use and waste compute resources in the type conversions,
//     defeating the purpose of the optimization in the first place.
//
// Note that this type is only available when compiling with Go 1.18 or later.
type GenericWriter[T any] struct {
	// At this time GenericWriter is expressed in terms of Writer to reuse the
	// underlying logic. In the future, and if we accepted to break backward
	// compatibility on the Write method, we could modify Writer to be an alias
	// to GenericWriter with:
	//
	//	type Writer = GenericWriter[any]
	//
	base Writer
	// This function writes rows of type T to the writer, it gets generated by
	// the NewGenericWriter function based on the type T and the underlying
	// schema of the parquet file.
	write writeFunc[T]
	// This field is used to leverage the optimized writeRowsFunc algorithms.
	columns []ColumnBuffer
}

// NewGenericWriter is like NewWriter but returns a GenericWriter[T] suited to
// write rows of Go type T.
//
// The type parameter T should be a map, struct, or any. Any other types will
// cause a panic at runtime. Type checking is a lot more effective when the
// generic parameter is a struct type, using map and interface types is somewhat
// similar to using a Writer.
//
// If the option list may explicitly declare a schema, it must be compatible
// with the schema generated from T.
//
// Sorting columns may be set on the writer to configure the generated row
// groups metadata. However, rows are always written in the order they were
// seen, no reordering is performed, the writer expects the application to
// ensure proper correlation between the order of rows and the list of sorting
// columns. See SortingWriter[T] for a writer which handles reordering rows
// based on the configured sorting columns.
func NewGenericWriter[T any](output io.Writer, options ...WriterOption) *GenericWriter[T] {
	config, err := NewWriterConfig(options...)
	if err != nil {
		panic(err)
	}

	t := typeOf[T]()

	var genWriteErr error
	if t != nil {
		if columnName, ok := validateColumns(dereference(t)); !ok {
			genWriteErr = fmt.Errorf("cannot write %v: it has columns with the same parquet column name %q", t, columnName)
		}
	}

	schemaFromConf := config.Schema
	schemaFromType := (*Schema)(nil)

	if t != nil && dereference(t).Kind() == reflect.Struct {
		schemaFromType = schemaOf(dereference(t), config.SchemaConfig.StructTags...)
	}

	config.Schema = cmp.Or(schemaFromConf, schemaFromType)
	if config.Schema == nil {
		panic("generic writer must be instantiated with schema or concrete type.")
	}
	if len(config.Schema.Columns()) == 0 {
		genWriteErr = fmt.Errorf("cannot write %v: it has no columns (maybe it has no exported fields)", t)
	}

	var writeFn writeFunc[T]
	switch {
	case genWriteErr != nil:
		writeFn = func(*GenericWriter[T], []T) (int, error) { return 0, genWriteErr }
	case schemaFromType == config.Schema || (schemaFromType != nil && EqualNodes(config.Schema, schemaFromType)):
		// The schema matches the type T, we can use the optimized
		// writeRowsFunc algorithms mapping Go values directly to
		// parquet columns, using the sparse package.
		writeRows := writeRowsFuncOf(t, config.Schema, nil, config.SchemaConfig.StructTags)
		writeFn = makeWriteFunc[T](t, writeRows)
	default:
		// The schema does not match the type T, we have to
		// deconstruct each value of type T into a Row first.
		// This is less efficient but still type-safe.
		_, writeValue := writeValueFuncOf(0, config.Schema)
		writeFn = makeWriteFunc[T](t, func(columns []ColumnBuffer, levels columnLevels, rows sparse.Array) {
			if rows.Len() == 0 {
				writeValue(columns, levels, reflect.Value{})
				return
			}
			for i := range rows.Len() {
				v := reflect.ValueOf((*T)(rows.Index(i))).Elem()
				writeValue(columns, levels, v)
			}
		})
	}

	return &GenericWriter[T]{
		base: Writer{
			output: output,
			config: config,
			schema: config.Schema,
			writer: newWriter(output, config),
		},
		write: writeFn,
	}
}

func validateColumns(t reflect.Type) (string, bool) {
	if t.Kind() != reflect.Struct {
		return "", true
	}

	columns := make(map[string]struct{}, t.NumField())

	for i := range t.NumField() {
		f := t.Field(i)
		if !f.IsExported() {
			continue
		}
		fieldTag, columnName := f.Tag.Get("parquet"), f.Name
		if fieldTag != "" {
			if commaIdx := strings.IndexByte(fieldTag, ','); commaIdx >= 0 {
				fieldTag = fieldTag[:commaIdx]
			}
			if fieldTag == "-" {
				continue
			}
			if fieldTag != "" {
				columnName = fieldTag
			}
		}
		if _, exists := columns[columnName]; exists {
			return columnName, false
		} else {
			columns[columnName] = struct{}{}
		}
	}

	return "", true
}

type writeFunc[T any] func(*GenericWriter[T], []T) (int, error)

func makeWriteFunc[T any](t reflect.Type, writeRows writeRowsFunc) writeFunc[T] {
	return func(w *GenericWriter[T], rows []T) (n int, err error) {
		if w.columns == nil {
			w.columns = make([]ColumnBuffer, len(w.base.writer.currentRowGroup.columns))
			for i, c := range w.base.writer.currentRowGroup.columns {
				// These fields are usually lazily initialized when writing rows,
				// we need them to exist now tho.
				c.columnBuffer = c.newColumnBuffer()
				w.columns[i] = c.columnBuffer
			}
		}
		writeRows(w.columns, columnLevels{}, makeArrayFromSlice(rows))
		return len(rows), nil
	}
}

// Close must be called after all values were produced to the writer in order to
// flush all buffers and write the parquet footer. The writer can only be reused
// if Reset is called first. Failure to do so will result in defined behavior.
func (w *GenericWriter[T]) Close() error {
	if err := w.base.Close(); err != nil {
		return err
	}
	return nil
}

func (w *GenericWriter[T]) Flush() error {
	return w.base.Flush()
}

// Reset clears the state of the writer without flushing any of the buffers,
// and setting the output to the io.Writer passed as argument, allowing the
// writer to be reused to produce another parquet file.
func (w *GenericWriter[T]) Reset(output io.Writer) {
	w.base.Reset(output)
}

func (w *GenericWriter[T]) Write(rows []T) (written int, err error) {
	var n int
	currentRowGroup := w.base.writer.currentRowGroup
	for len(rows) > 0 {
		n, err = currentRowGroup.writeRows(len(rows), func(i, j int) (int, error) {
			n, err := w.write(w, rows[i:j:j])
			if err != nil {
				return n, err
			}

			for _, c := range currentRowGroup.columns {
				if c.columnBuffer != nil && c.columnBuffer.Size() >= int64(c.bufferSize) {
					if err := c.Flush(); err != nil {
						return n, err
					}
				}
			}

			return n, nil
		})
		rows = rows[n:]
		written += n
		if err != ErrTooManyRowGroups {
			break
		}
		if err = w.base.writer.flush(); err != nil {
			break
		}
	}
	return
}

func (w *GenericWriter[T]) WriteRows(rows []Row) (int, error) {
	return w.base.WriteRows(rows)
}

func (w *GenericWriter[T]) WriteRowGroup(rowGroup RowGroup) (int64, error) {
	return w.base.WriteRowGroup(rowGroup)
}

// SetKeyValueMetadata sets a key/value pair in the Parquet file metadata.
//
// Keys are assumed to be unique, if the same key is repeated multiple times the
// last value is retained. While the parquet format does not require unique keys,
// this design decision was made to optimize for the most common use case where
// applications leverage this extension mechanism to associate single values to
// keys. This may create incompatibilities with other parquet libraries, or may
// cause some key/value pairs to be lost when open parquet files written with
// repeated keys. We can revisit this decision if it ever becomes a blocker.
func (w *GenericWriter[T]) SetKeyValueMetadata(key, value string) {
	w.base.SetKeyValueMetadata(key, value)
}

func (w *GenericWriter[T]) ReadRowsFrom(rows RowReader) (int64, error) {
	return w.base.ReadRowsFrom(rows)
}

func (w *GenericWriter[T]) Schema() *Schema {
	return w.base.Schema()
}

// Size returns an estimate of the current file size in bytes.
// See Writer.Size for details.
func (w *GenericWriter[T]) Size() int64 {
	return w.base.Size()
}

func (w *GenericWriter[T]) ColumnWriters() []*ColumnWriter {
	return w.base.ColumnWriters()
}

// File returns a FileView of the written parquet file.
// Only available after Close is called.
func (w *GenericWriter[T]) File() FileView {
	return w.base.File()
}

// ConcurrentRowGroupWriter is a row group writer that can be used to write row groups
// in parallel. Multiple row groups can be created concurrently and written to independently,
// but they must be committed serially to maintain the order of row groups in the file.
//
// See BeginRowGroup for more information on how this can be used.
//
// While multiple row groups can be created concurrently, a single row group must be written
// sequentially.
type ConcurrentRowGroupWriter struct {
	writer      *writer
	config      *WriterConfig
	values      [][]Value
	numRows     int64
	maxRows     int64
	columns     []*ColumnWriter
	columnChunk []format.ColumnChunk
	columnIndex []format.ColumnIndex
	offsetIndex []format.OffsetIndex
}

// BeginRowGroup returns a new ConcurrentRowGroupWriter that can be written to in parallel with
// other row groups. However these need to be committed back to the writer serially using the
// Commit method on the row group.
//
// Example usage could look something like:
//
//	writer := parquet.NewGenericWriter[any](...)
//	rgs := make([]*parquet.ConcurrentRowGroupWriter, 5)
//	var wg sync.WaitGroup
//	for i := range rgs {
//	  rg := writer.BeginRowGroup()
//	  rgs[i] = rg
//	  wg.Add(1)
//	  go func() {
//	    defer wg.Done()
//	    writeChunkRows(i, rg)
//	  }()
//	}
//	wg.Wait()
//	for _, rg := range rgs {
//	  if _, err := rg.Commit(); err != nil {
//	    return err
//	  }
//	}
//	return writer.Close()
func (w *GenericWriter[T]) BeginRowGroup() *ConcurrentRowGroupWriter {
	return newConcurrentRowGroupWriter(w.base.writer, w.base.config)
}

var (
	_ RowWriterWithSchema = (*GenericWriter[any])(nil)
	_ RowReaderFrom       = (*GenericWriter[any])(nil)
	_ RowGroupWriter      = (*GenericWriter[any])(nil)

	_ RowWriterWithSchema = (*GenericWriter[struct{}])(nil)
	_ RowReaderFrom       = (*GenericWriter[struct{}])(nil)
	_ RowGroupWriter      = (*GenericWriter[struct{}])(nil)

	_ RowWriterWithSchema = (*GenericWriter[map[struct{}]struct{}])(nil)
	_ RowReaderFrom       = (*GenericWriter[map[struct{}]struct{}])(nil)
	_ RowGroupWriter      = (*GenericWriter[map[struct{}]struct{}])(nil)
)

// Deprecated: A Writer uses a parquet schema and sequence of Go values to
// produce a parquet file to an io.Writer.
//
// Use NewGenericWriter instead. To maintain dynamic behavior (schema unknown at compile time),
// use "any" as the type parameter:
//
//	This gives you the same behavior as the old Writer
//	w := parquet.NewGenericWriter[any](output, schema)
//
// This example showcases a typical use of parquet writers:
//
//	writer := parquet.NewWriter(output)
//
//	for _, row := range rows {
//		if err := writer.Write(row); err != nil {
//			...
//		}
//	}
//
//	if err := writer.Close(); err != nil {
//		...
//	}
//
// The Writer type optimizes for minimal memory usage, each page is written as
// soon as it has been filled so only a single page per column needs to be held
// in memory and as a result, there are no opportunities to sort rows within an
// entire row group. Programs that need to produce parquet files with sorted
// row groups should use the Buffer type to buffer and sort the rows prior to
// writing them to a Writer.
//
// For programs building with Go 1.18 or later, the GenericWriter[T] type
// supersedes this one.
type Writer struct {
	output io.Writer
	config *WriterConfig
	schema *Schema
	writer *writer
	rowbuf []Row
}

// NewWriter constructs a parquet writer writing a file to the given io.Writer.
//
// The function panics if the writer configuration is invalid. Programs that
// cannot guarantee the validity of the options passed to NewWriter should
// construct the writer configuration independently prior to calling this
// function:
//
//	config, err := parquet.NewWriterConfig(options...)
//	if err != nil {
//		// handle the configuration error
//		...
//	} else {
//		// this call to create a writer is guaranteed not to panic
//		writer := parquet.NewWriter(output, config)
//		...
//	}
func NewWriter(output io.Writer, options ...WriterOption) *Writer {
	config, err := NewWriterConfig(options...)
	if err != nil {
		panic(err)
	}
	w := &Writer{
		output: output,
		config: config,
	}
	if config.Schema != nil {
		w.configure(config.Schema)
	}
	return w
}

func (w *Writer) configure(schema *Schema) {
	if schema != nil {
		w.config.Schema = schema
		w.schema = schema
		w.writer = newWriter(w.output, w.config)
	}
}

// Close must be called after all values were produced to the writer in order to
// flush all buffers and write the parquet footer.
func (w *Writer) Close() error {
	for _, c := range w.ColumnWriters() {
		if err := c.Close(); err != nil {
			return err
		}
	}
	if w.writer != nil {
		return w.writer.close()
	}
	return nil
}

// Flush flushes all buffers into a row group to the underlying io.Writer.
//
// Flush is called automatically on Close, it is only useful to call explicitly
// if the application needs to limit the size of row groups or wants to produce
// multiple row groups per file.
//
// If the writer attempts to create more than MaxRowGroups row groups the method
// returns ErrTooManyRowGroups.
func (w *Writer) Flush() error {
	if w.writer != nil {
		return w.writer.flush()
	}
	return nil
}

// Reset clears the state of the writer without flushing any of the buffers,
// and setting the output to the io.Writer passed as argument, allowing the
// writer to be reused to produce another parquet file.
//
// Reset may be called at any time, including after a writer was closed.
func (w *Writer) Reset(output io.Writer) {
	if w.output = output; w.writer != nil {
		w.writer.reset(w.output)
	}
}

// Write is called to write another row to the parquet file.
//
// The method uses the parquet schema configured on w to traverse the Go value
// and decompose it into a set of columns and values. If no schema were passed
// to NewWriter, it is deducted from the Go type of the row, which then have to
// be a struct or pointer to struct.
func (w *Writer) Write(row any) error {
	if w.schema == nil {
		w.configure(SchemaOf(row))
	}
	if cap(w.rowbuf) == 0 {
		w.rowbuf = make([]Row, 1)
	} else {
		w.rowbuf = w.rowbuf[:1]
	}
	defer clearRows(w.rowbuf)
	w.rowbuf[0] = w.schema.Deconstruct(w.rowbuf[0][:0], row)
	_, err := w.WriteRows(w.rowbuf)
	return err
}

// WriteRows is called to write rows to the parquet file.
//
// The Writer must have been given a schema when NewWriter was called, otherwise
// the structure of the parquet file cannot be determined from the row only.
//
// The row is expected to contain values for each column of the writer's schema,
// in the order produced by the parquet.(*Schema).Deconstruct method.
func (w *Writer) WriteRows(rows []Row) (int, error) {
	return w.writer.WriteRows(rows)
}

// WriteRowGroup writes a row group to the parquet file.
//
// Buffered rows will be flushed prior to writing rows from the group, unless
// the row group was empty in which case nothing is written to the file.
//
// The content of the row group is flushed to the writer; after the method
// returns successfully, the row group will be empty and in ready to be reused.
func (w *Writer) WriteRowGroup(rowGroup RowGroup) (int64, error) {
	rowGroupSchema := rowGroup.Schema()
	switch {
	case rowGroupSchema == nil:
		return 0, ErrRowGroupSchemaMissing
	case w.schema == nil:
		w.configure(rowGroupSchema)
	case !EqualNodes(w.schema, rowGroupSchema):
		return 0, ErrRowGroupSchemaMismatch
	}
	if err := w.writer.flush(); err != nil {
		return 0, err
	}
	w.writer.currentRowGroup.configureBloomFilters(rowGroup.ColumnChunks())
	rows := rowGroup.Rows()
	defer rows.Close()
	n, err := CopyRows(w.writer, rows)
	if err != nil {
		return n, err
	}
	return w.writer.writeRowGroup(w.writer.currentRowGroup, rowGroup.Schema(), rowGroup.SortingColumns())
}

// ReadRowsFrom reads rows from the reader passed as arguments and writes them
// to w.
//
// This is similar to calling WriteRow repeatedly, but will be more efficient
// if optimizations are supported by the reader.
func (w *Writer) ReadRowsFrom(rows RowReader) (written int64, err error) {
	if w.schema == nil {
		if r, ok := rows.(RowReaderWithSchema); ok {
			w.configure(r.Schema())
		}
	}
	if cap(w.rowbuf) < defaultRowBufferSize {
		w.rowbuf = make([]Row, defaultRowBufferSize)
	} else {
		w.rowbuf = w.rowbuf[:cap(w.rowbuf)]
	}
	return copyRows(w.writer, rows, w.rowbuf)
}

// Schema returns the schema of rows written by w.
//
// The returned value will be nil if no schema has yet been configured on w.
func (w *Writer) Schema() *Schema { return w.schema }

// Size returns an estimate of the current file size in bytes, including all
// completed row groups and the current in-progress row group.
//
// The estimate includes bytes already written to the underlying writer plus
// the estimated size of buffered data in the current row group. The buffered
// data has not been compressed yet, so this is an upper-bound estimate.
//
// The size of the footer metadata is not included because it is variable and
// typically small relative to the data.
//
// This method can be used to target a specific file size by calling Flush
// when Size reaches a threshold.
func (w *Writer) Size() int64 {
	if w.writer == nil {
		return 0
	}
	return w.writer.writer.offset + w.writer.currentRowGroup.Size()
}

// SetKeyValueMetadata sets a key/value pair in the Parquet file metadata.
//
// Keys are assumed to be unique, if the same key is repeated multiple times the
// last value is retained. While the parquet format does not require unique keys,
// this design decision was made to optimize for the most common use case where
// applications leverage this extension mechanism to associate single values to
// keys. This may create incompatibilities with other parquet libraries, or may
// cause some key/value pairs to be lost when open parquet files written with
// repeated keys. We can revisit this decision if it ever becomes a blocker.
func (w *Writer) SetKeyValueMetadata(key, value string) {
	for i, kv := range w.writer.metadata {
		if kv.Key == key {
			kv.Value = value
			w.writer.metadata[i] = kv
			return
		}
	}
	w.writer.metadata = append(w.writer.metadata, format.KeyValue{
		Key:   key,
		Value: value,
	})
}

// ColumnWriters returns writers for each column. This allows applications to
// write values directly to each column instead of having to first assemble
// values into rows to use WriteRows.
func (w *Writer) ColumnWriters() []*ColumnWriter { return w.writer.currentRowGroup.columns }

// BeginRowGroup returns a new ConcurrentRowGroupWriter that can be written to in parallel with
// other row groups. However these need to be committed back to the writer serially using the
// Commit method on the row group.
func (w *Writer) BeginRowGroup() *ConcurrentRowGroupWriter {
	return newConcurrentRowGroupWriter(w.writer, w.config)
}

type writerFileView struct {
	writer *writer
	schema *Schema
}

// File returns a FileView of the written parquet file.
// Only available after Close is called.
func (w *Writer) File() FileView {
	if w.writer == nil || w.schema == nil {
		return nil
	}
	return &writerFileView{
		w.writer,
		w.schema,
	}
}

func (w *writerFileView) Metadata() *format.FileMetaData {
	return &w.writer.fileMetaData
}

func (w *writerFileView) Schema() *Schema {
	return w.schema
}

func (w *writerFileView) NumRows() int64 {
	return w.writer.fileMetaData.NumRows
}

func (w *writerFileView) Lookup(key string) (string, bool) {
	return lookupKeyValueMetadata(w.writer.fileMetaData.KeyValueMetadata, key)
}

func (w *writerFileView) Size() int64 {
	return w.writer.writer.offset
}

func (w *writerFileView) ColumnIndexes() []format.ColumnIndex {
	return w.writer.currentRowGroup.columnIndex
}

func (w *writerFileView) OffsetIndexes() []format.OffsetIndex {
	return w.writer.currentRowGroup.offsetIndex
}

func (w *writerFileView) Root() *Column {
	root, _ := openColumns(nil, &w.writer.fileMetaData, w.writer.currentRowGroup.columnIndex, w.writer.currentRowGroup.offsetIndex)
	return root
}

func (w *writerFileView) RowGroups() []RowGroup {
	columns := makeLeafColumns(w.Root())
	file := &File{metadata: w.writer.fileMetaData, schema: w.schema}
	fileRowGroups := makeFileRowGroups(file, columns)
	return makeRowGroups(fileRowGroups)
}

func newConcurrentRowGroupWriter(w *writer, config *WriterConfig) *ConcurrentRowGroupWriter {
	rg := &ConcurrentRowGroupWriter{
		writer:  w,
		config:  config,
		maxRows: config.MaxRowsPerRowGroup,
	}

	dataPageType := format.DataPage
	if config.DataPageVersion == 2 {
		dataPageType = format.DataPageV2
	}

	defaultCompression := config.Compression
	if defaultCompression == nil {
		defaultCompression = &Uncompressed
	}

	forEachLeafColumnOf(config.Schema, func(leaf leafColumn) {
		encoding := encodingOf(leaf.node, config.Encodings)
		dictionary := Dictionary(nil)
		columnType := leaf.node.Type()
		columnIndex := int(leaf.columnIndex)
		compression := leaf.node.Compression()

		if compression == nil {
			compression = defaultCompression
		}

		if isDictionaryEncoding(encoding) {
			dictBuffer := columnType.NewValues(make([]byte, 0, defaultDictBufferSize), nil)
			dictionary = columnType.NewDictionary(columnIndex, 0, dictBuffer)
			columnType = dictionary.Type()
		}

		c := &ColumnWriter{
			pool:               config.ColumnPageBuffers,
			columnPath:         leaf.path,
			columnType:         columnType,
			originalType:       columnType,
			columnIndex:        columnType.NewColumnIndexer(config.ColumnIndexSizeLimit(leaf.path)),
			columnFilter:       searchBloomFilterColumn(config.BloomFilters, leaf.path),
			compression:        compression,
			dictionary:         dictionary,
			maxRepetitionLevel: leaf.maxRepetitionLevel,
			maxDefinitionLevel: leaf.maxDefinitionLevel,
			bufferIndex:        int32(leaf.columnIndex),
			bufferSize:         int32(float64(config.PageBufferSize) * 0.98),
			writePageStats: config.DataPageStatistics && !slices.ContainsFunc(config.SkipPageStatistics, func(skip []string) bool {
				return columnPath(skip).equal(leaf.path)
			}),
			writePageBounds: !slices.ContainsFunc(config.SkipPageBounds, func(skip []string) bool {
				return columnPath(skip).equal(leaf.path)
			}),
			writeDeprecatedStatistics: config.DeprecatedDataPageStatistics,
			encodings:                 make([]format.Encoding, 0, 3),
			// Data pages in version 2 can omit compression when dictionary
			// encoding is employed; only the dictionary page needs to be
			// compressed, the data pages are encoded with the hybrid
			// RLE/Bit-Pack encoding which doesn't benefit from an extra
			// compression layer.
			isCompressed:       isCompressed(compression) && (dataPageType != format.DataPageV2 || dictionary == nil),
			dictionaryMaxBytes: config.DictionaryMaxBytes,
		}

		if dictionary != nil {
			c.header.dict.Type = format.DictionaryPage
			c.header.dict.DictionaryPageHeader.Valid = true
		}

		c.header.page.Type = dataPageType
		switch dataPageType {
		case format.DataPage:
			c.header.page.DataPageHeader.Valid = true
		case format.DataPageV2:
			c.header.page.DataPageHeaderV2.Valid = true
		}

		c.header.encoder.Reset(c.header.protocol.NewWriter(&c.header.buffer))

		if c.maxRepetitionLevel > 0 {
			c.repetitionLevelHistogram = make([]int64, int(c.maxRepetitionLevel)+1)
		}
		if c.maxDefinitionLevel > 0 {
			c.definitionLevelHistogram = make([]int64, int(c.maxDefinitionLevel)+1)
			c.encodings = addEncoding(c.encodings, format.RLE)
		}

		if isDictionaryEncoding(encoding) {
			c.encodings = addEncoding(c.encodings, format.Plain)
		}

		c.encoding = encoding
		c.originalEncoding = encoding
		c.encodings = addEncoding(c.encodings, c.encoding.Encoding())
		sortPageEncodings(c.encodings)

		rg.columns = append(rg.columns, c)
	})

	// Pre-allocate the backing array so that in most cases where the rows
	// contain a single value we will hit collocated memory areas when writing
	// rows to the writer. This won't benefit repeated columns much but in that
	// case we would just waste a bit of memory which we can afford.
	values := make([]Value, len(rg.columns))
	rg.values = make([][]Value, len(rg.columns))
	for i := range values {
		rg.values[i] = values[i : i : i+1]
	}

	rg.columnChunk = make([]format.ColumnChunk, len(rg.columns))
	rg.columnIndex = make([]format.ColumnIndex, len(rg.columns))
	rg.offsetIndex = make([]format.OffsetIndex, len(rg.columns))

	for i, c := range rg.columns {
		rg.columnChunk[i] = format.ColumnChunk{
			MetaData: format.ColumnMetaData{
				Type:             format.Type(c.columnType.Kind()),
				Encoding:         c.encodings,
				PathInSchema:     thrift.Slice[string](c.columnPath),
				Codec:            c.compression.CompressionCodec(),
				KeyValueMetadata: nil, // TODO
			},
		}
	}

	for i, c := range rg.columns {
		c.columnChunk = &rg.columnChunk[i]
		c.offsetIndex = &rg.offsetIndex[i]
	}

	return rg
}

func (rg *ConcurrentRowGroupWriter) reset() {
	rg.numRows = 0
	for _, c := range rg.columns {
		c.reset()
	}
}

func (rg *ConcurrentRowGroupWriter) configureBloomFilters(columnChunks []ColumnChunk) {
	for i, c := range rg.columns {
		if c.columnFilter != nil {
			c.resizeBloomFilter(columnChunks[i].NumValues())
		}
	}
}

// Schema returns the schema for this row group.
func (rg *ConcurrentRowGroupWriter) Schema() *Schema {
	return rg.config.Schema
}

// ColumnWriters returns the column writers for this row group, allowing
// direct access to write values to individual columns.
func (rg *ConcurrentRowGroupWriter) ColumnWriters() []*ColumnWriter {
	return rg.columns
}

// Flush flushes any buffered data in the row group's column writers.
// This could be called before Commit to ensure all data pages are flushed.
func (rg *ConcurrentRowGroupWriter) Flush() error {
	for _, c := range rg.columns {
		if err := c.Flush(); err != nil {
			return err
		}
	}
	return nil
}

// Size returns an estimate of the current row group size in bytes.
//
// The estimate sums encoded pages, buffered values not yet encoded, dictionary
// pages, and bloom filters for each column. Because buffered values have not
// been compressed, this is an upper-bound estimate.
func (rg *ConcurrentRowGroupWriter) Size() int64 {
	size := int64(0)
	for _, c := range rg.columns {
		size += c.columnChunk.MetaData.TotalCompressedSize
		if c.columnBuffer != nil {
			size += c.columnBuffer.Size()
		}
		if c.dictionary != nil {
			size += c.dictionary.Size()
		}
		size += int64(len(c.filter))
	}
	return size
}

// Commit commits the row group to the parent writer, returning the number
// of rows written and an error if any. This method must be called serially
// (not concurrently) to maintain row group order in the file.
//
// If the parent writer has any pending rows buffered, they will be flushed
// before this row group is written.
//
// After Commit returns successfully, the row group will be empty and can
// be reused.
func (rg *ConcurrentRowGroupWriter) Commit() (int64, error) {
	if err := rg.writer.flush(); err != nil {
		return 0, err
	}
	return rg.writer.writeRowGroup(rg, nil, nil)
}

// WriteRows writes rows to the row group.
func (rg *ConcurrentRowGroupWriter) WriteRows(rows []Row) (int, error) {
	return rg.writeRows(len(rows), func(start, end int) (int, error) {
		defer func() {
			for i, values := range rg.values {
				clearValues(values)
				rg.values[i] = values[:0]
			}
		}()

		// TODO: if an error occurs in this method the writer may be left in an
		// partially functional state. Applications are not expected to continue
		// using the writer after getting an error, but maybe we could ensure that
		// we are preventing further use as well?
		for _, row := range rows[start:end] {
			for columnIndex, columnValues := range row.Range {
				rg.values[columnIndex] = append(rg.values[columnIndex], columnValues...)
			}
		}

		for i, values := range rg.values {
			if len(values) > 0 {
				if _, err := rg.columns[i].WriteRowValues(values); err != nil {
					return 0, err
				}
			}
		}

		return end - start, nil
	})
}

func (rg *ConcurrentRowGroupWriter) writeRows(numRows int, write func(i, j int) (int, error)) (int, error) {
	written := 0

	for written < numRows {
		remain := rg.maxRows - rg.numRows
		length := numRows - written

		if remain <= 0 {
			return written, ErrTooManyRowGroups
		}

		if remain < int64(length) {
			length = int(remain)
		}

		// Since the writer cannot flush pages across row boundaries, calls to
		// WriteRows with very large slices can result in greatly exceeding the
		// target page size. To set a limit to the impact of these large writes
		// we chunk the input in slices of 64 rows.
		const maxRowsPerWrite = 64
		if length > maxRowsPerWrite {
			length = maxRowsPerWrite
		}

		n, err := write(written, written+length)
		written += n
		rg.numRows += int64(n)
		if err != nil {
			return written, err
		}
	}

	return written, nil
}

type writer struct {
	buffer          *bufio.Writer
	writer          offsetTrackingWriter
	currentRowGroup *ConcurrentRowGroupWriter

	createdBy string
	metadata  []format.KeyValue

	columnOrders   []format.ColumnOrder
	schemaElements []format.SchemaElement
	rowGroups      []format.RowGroup
	columnIndexes  [][]format.ColumnIndex
	offsetIndexes  [][]format.OffsetIndex
	sortingColumns []format.SortingColumn

	fileMetaData format.FileMetaData
	footer       [8]byte
}

func newWriter(output io.Writer, config *WriterConfig) *writer {
	w := new(writer)
	if config.WriteBufferSize <= 0 {
		w.writer.Reset(output)
	} else {
		w.buffer = bufio.NewWriterSize(output, config.WriteBufferSize)
		w.writer.Reset(w.buffer)
	}
	w.createdBy = config.CreatedBy
	w.metadata = make([]format.KeyValue, 0, len(config.KeyValueMetadata))
	for k, v := range config.KeyValueMetadata {
		w.metadata = append(w.metadata, format.KeyValue{Key: k, Value: v})
	}
	sortKeyValueMetadata(w.metadata)
	w.sortingColumns = make([]format.SortingColumn, len(config.Sorting.SortingColumns))

	config.Schema.forEachNode(func(name string, node Node) {
		nodeType := node.Type()

		elem := format.SchemaElement{
			Name:    name,
			FieldID: int32(node.ID()),
		}

		if node != config.Schema { // the root has no repetition type
			elem.RepetitionType = fieldRepetitionTypeNullOf(node)
		}

		// For backward compatibility with older readers, the parquet specification
		// recommends to set the scale and precision on schema elements when the
		// column is of logical type decimal.
		logicalType := nodeType.LogicalType()
		if logicalType != nil {
			elem.LogicalType.Set(*logicalType)
			if logicalType.Decimal != nil {
				elem.Scale.Set(logicalType.Decimal.Scale)
				elem.Precision.Set(logicalType.Decimal.Precision)
			}
		}

		if n := int32(nodeType.Length()); n > 0 {
			elem.TypeLength.Set(n)
		}

		if !node.Leaf() {
			n := int32(len(node.Fields()))
			elem.NumChildren.Set(n)
		}

		if physType := nodeType.PhysicalType(); physType != nil {
			elem.Type.Set(*physType)
		}

		if convType := nodeType.ConvertedType(); convType != nil {
			elem.ConvertedType.Set(*convType)
		}

		w.schemaElements = append(w.schemaElements, elem)
	})

	w.currentRowGroup = newConcurrentRowGroupWriter(w, config)

	if len(config.Sorting.SortingColumns) > 0 {
		forEachLeafColumnOf(config.Schema, func(leaf leafColumn) {
			if sortingIndex := searchSortingColumn(config.Sorting.SortingColumns, leaf.path); sortingIndex < len(w.sortingColumns) {
				w.sortingColumns[sortingIndex] = format.SortingColumn{
					ColumnIdx:  int32(leaf.columnIndex),
					Descending: config.Sorting.SortingColumns[sortingIndex].Descending(),
					NullsFirst: config.Sorting.SortingColumns[sortingIndex].NullsFirst(),
				}
			}
		})
	}

	w.columnOrders = make([]format.ColumnOrder, len(w.currentRowGroup.columns))
	for i, c := range w.currentRowGroup.columns {
		w.columnOrders[i] = *c.columnType.ColumnOrder()
	}

	copy(w.footer[4:], "PAR1")
	return w
}

func (w *writer) reset(writer io.Writer) {
	if w.buffer == nil {
		w.writer.Reset(writer)
	} else {
		w.buffer.Reset(writer)
		w.writer.Reset(w.buffer)
	}
	w.currentRowGroup.reset()
	for i := range w.rowGroups {
		w.rowGroups[i].Reset()
	}
	for i := range w.columnIndexes {
		for j := range w.columnIndexes[i] {
			w.columnIndexes[i][j].Reset()
		}
		w.columnIndexes[i] = w.columnIndexes[i][:0]
	}
	for i := range w.offsetIndexes {
		for j := range w.offsetIndexes[i] {
			w.offsetIndexes[i][j].Reset()
		}
		w.offsetIndexes[i] = w.offsetIndexes[i][:0]
	}
	w.rowGroups = w.rowGroups[:0]
	w.columnIndexes = w.columnIndexes[:0]
	w.offsetIndexes = w.offsetIndexes[:0]
	w.fileMetaData = format.FileMetaData{}
}

func (w *writer) close() error {
	if err := w.writeFileHeader(); err != nil {
		return err
	}
	if err := w.flush(); err != nil {
		return err
	}
	if err := w.writeFileFooter(); err != nil {
		return err
	}
	if w.buffer != nil {
		return w.buffer.Flush()
	}
	return nil
}

func (w *writer) flush() error {
	_, err := w.writeRowGroup(w.currentRowGroup, nil, nil)
	return err
}

func (w *writer) writeFileHeader() error {
	if w.writer.writer == nil {
		return io.ErrClosedPipe
	}
	if w.writer.offset == 0 {
		_, err := w.writer.WriteString("PAR1")
		return err
	}
	return nil
}

func (w *writer) writeFileFooter() error {
	// The page index is composed of two sections: column and offset indexes.
	// They are written after the row groups, right before the footer (which
	// is written by the parent Writer.Close call).
	//
	// This section both writes the page index and generates the values of
	// ColumnIndexOffset, ColumnIndexLength, OffsetIndexOffset, and
	// OffsetIndexLength in the corresponding columns of the file metadata.
	//
	// Note: the page index is always written, even if we created data pages v1
	// because the parquet format is backward compatible in this case. Older
	// readers will simply ignore this section since they do not know how to
	// decode its content, nor have loaded any metadata to reference it.
	protocol := new(thrift.CompactProtocol)
	encoder := thrift.NewEncoder(protocol.NewWriter(&w.writer))

	for i, columnIndexes := range w.columnIndexes {
		rowGroup := &w.rowGroups[i]
		for j := range columnIndexes {
			column := &rowGroup.Columns[j]
			column.ColumnIndexOffset = w.writer.offset
			if err := encoder.Encode(&columnIndexes[j]); err != nil {
				return err
			}
			column.ColumnIndexLength = int32(w.writer.offset - column.ColumnIndexOffset)
		}
	}

	for i, offsetIndexes := range w.offsetIndexes {
		rowGroup := &w.rowGroups[i]
		for j := range offsetIndexes {
			column := &rowGroup.Columns[j]
			column.OffsetIndexOffset = w.writer.offset
			if err := encoder.Encode(&offsetIndexes[j]); err != nil {
				return err
			}
			column.OffsetIndexLength = int32(w.writer.offset - column.OffsetIndexOffset)
		}
	}

	numRows := int64(0)
	for rowGroupIndex := range w.rowGroups {
		numRows += w.rowGroups[rowGroupIndex].NumRows
	}

	// We implemented the parquet specification version 2+, which is represented
	// by the version number 2 in the file metadata.
	//
	// For reference, see:
	// https://github.com/apache/arrow/blob/70b9ef5/go/parquet/metadata/file.go#L122-L127
	const parquetFileFormatVersion = 2

	w.fileMetaData = format.FileMetaData{
		Version:          parquetFileFormatVersion,
		Schema:           w.schemaElements,
		NumRows:          numRows,
		RowGroups:        w.rowGroups,
		KeyValueMetadata: w.metadata,
		CreatedBy:        w.createdBy,
		ColumnOrders:     w.columnOrders,
	}

	length := w.writer.offset
	if err := encoder.Encode(&w.fileMetaData); err != nil {
		return err
	}

	length = w.writer.offset - length
	binary.LittleEndian.PutUint32(w.footer[:4], uint32(length))

	_, err := w.writer.Write(w.footer[:])
	return err
}

func (w *writer) writeRowGroup(rg *ConcurrentRowGroupWriter, rowGroupSchema *Schema, rowGroupSortingColumns []SortingColumn) (int64, error) {
	if len(rg.columns) == 0 {
		return 0, nil
	}
	numRows := rg.columns[0].totalRowCount()
	if numRows == 0 {
		return 0, nil
	}

	rowGroupIndex := len(w.rowGroups)
	columnIndexIndex := len(w.columnIndexes)
	offsetIndexIndex := len(w.offsetIndexes)

	if len(w.rowGroups) == MaxRowGroups {
		return 0, ErrTooManyRowGroups
	}

	defer func() {
		rg.reset()
	}()

	for _, c := range rg.columns {
		if err := c.Flush(); err != nil {
			return 0, err
		}
		if err := c.flushFilterPages(); err != nil {
			return 0, err
		}
	}

	if err := w.writeFileHeader(); err != nil {
		return 0, err
	}
	fileOffset := w.writer.offset

	for i, c := range rg.columns {
		rg.columnIndex[i] = c.columnIndex.ColumnIndex()
		rg.columnIndex[i].RepetitionLevelHistogram = append(rg.columnIndex[i].RepetitionLevelHistogram[:0], c.pageRepetitionLevelHistograms...)
		rg.columnIndex[i].DefinitionLevelHistogram = append(rg.columnIndex[i].DefinitionLevelHistogram[:0], c.pageDefinitionLevelHistograms...)

		c.columnChunk.MetaData.SizeStatistics = format.SizeStatistics{
			UnencodedByteArrayDataBytes: c.totalUnencodedByteArrayBytes,
			RepetitionLevelHistogram:    slices.Clone(c.repetitionLevelHistogram),
			DefinitionLevelHistogram:    slices.Clone(c.definitionLevelHistogram),
		}

		if c.dictionary != nil {
			c.columnChunk.MetaData.DictionaryPageOffset = w.writer.offset
			if err := c.writeDictionaryPage(&w.writer, c.dictionary); err != nil {
				return 0, fmt.Errorf("writing dictionary page of row group colum %d: %w", i, err)
			}
		}

		// Skip columns with nil pageBuffer (e.g., empty struct groups with no leaf columns)
		if c.pageBuffer == nil {
			continue
		}

		dataPageOffset := w.writer.offset
		c.columnChunk.MetaData.DataPageOffset = dataPageOffset
		for j := range c.offsetIndex.PageLocations {
			c.offsetIndex.PageLocations[j].Offset += dataPageOffset
		}

		if offset, err := c.pageBuffer.Seek(0, io.SeekStart); err != nil {
			return 0, err
		} else if offset != 0 {
			return 0, fmt.Errorf("resetting parquet page buffer to the start expected offset zero but got %d", offset)
		}
		if _, err := io.Copy(&w.writer, c.pageBuffer); err != nil {
			return 0, fmt.Errorf("writing buffered pages of row group column %d: %w", i, err)
		}
	}

	for _, c := range rg.columns {
		if len(c.filter) > 0 {
			bloomFilterOffset := w.writer.offset
			c.columnChunk.MetaData.BloomFilterOffset = bloomFilterOffset
			if err := c.writeBloomFilter(&w.writer); err != nil {
				return 0, err
			}
			bloomFilterLength := w.writer.offset - bloomFilterOffset
			c.columnChunk.MetaData.BloomFilterLength = int32(bloomFilterLength)
		}
	}

	totalByteSize := int64(0)
	totalCompressedSize := int64(0)

	for i := range rg.columnChunk {
		c := &rg.columnChunk[i].MetaData
		sortPageEncodingStats(c.EncodingStats)
		totalByteSize += int64(c.TotalUncompressedSize)
		totalCompressedSize += int64(c.TotalCompressedSize)
	}

	var reuseRowGroup *format.RowGroup = nil
	if cap(w.rowGroups) > rowGroupIndex {
		w.rowGroups = w.rowGroups[:rowGroupIndex+1] // Extend the slice by one element
		reuseRowGroup = &w.rowGroups[rowGroupIndex] // Pointer to the last element
	}

	sortingColumns := w.sortingColumns
	if len(sortingColumns) == 0 && len(rowGroupSortingColumns) > 0 {
		scLen := len(rowGroupSortingColumns)
		if reuseRowGroup == nil {
			sortingColumns = make([]format.SortingColumn, 0, scLen)
		} else {
			sortingColumns = reuseRowGroup.SortingColumns
		}
		forEachLeafColumnOf(rowGroupSchema, func(leaf leafColumn) {
			if sortingIndex := searchSortingColumn(rowGroupSortingColumns, leaf.path); sortingIndex < len(sortingColumns) {
				sortingColumns[sortingIndex] = format.SortingColumn{
					ColumnIdx:  int32(leaf.columnIndex),
					Descending: rowGroupSortingColumns[sortingIndex].Descending(),
					NullsFirst: rowGroupSortingColumns[sortingIndex].NullsFirst(),
				}
			}
		})
	}

	var reuseColumnIndex *[]format.ColumnIndex = nil
	if cap(w.columnIndexes) > columnIndexIndex {
		// Extend the slice by one element
		w.columnIndexes = w.columnIndexes[:columnIndexIndex+1]
		reuseColumnIndex = &w.columnIndexes[columnIndexIndex]
	}
	var columnIndex []format.ColumnIndex
	if reuseColumnIndex != nil {
		// Copy the slice
		columnIndex = append(*reuseColumnIndex, rg.columnIndex...)
		w.columnIndexes[columnIndexIndex] = columnIndex
	} else {
		columnIndex = slices.Clone(rg.columnIndex)
		w.columnIndexes = append(w.columnIndexes, columnIndex)
	}

	var reuseOffsetIndex *[]format.OffsetIndex = nil
	if cap(w.offsetIndexes) > offsetIndexIndex {
		// Extend the slice by one element
		w.offsetIndexes = w.offsetIndexes[:offsetIndexIndex+1]
		reuseOffsetIndex = &w.offsetIndexes[offsetIndexIndex]
	}
	var offsetIndex []format.OffsetIndex
	if reuseOffsetIndex != nil {
		offsetIndex = append(*reuseOffsetIndex, rg.offsetIndex...)
		w.offsetIndexes[offsetIndexIndex] = offsetIndex
	} else {
		offsetIndex = slices.Clone(rg.offsetIndex)
		w.offsetIndexes = append(w.offsetIndexes, offsetIndex)
	}

	// If the existing row group slice is full, we need to create a new element
	var columns []format.ColumnChunk = nil
	if reuseRowGroup == nil {
		columns = slices.Clone(rg.columnChunk)
	} else {
		columns = append(reuseRowGroup.Columns, rg.columnChunk...)
	}
	for i := range columns {
		c := &columns[i]
		c.MetaData.EncodingStats = append(c.MetaData.EncodingStats[:0], rg.columnChunk[i].MetaData.EncodingStats...)
	}

	for i := range offsetIndex {
		c := &offsetIndex[i]
		c.PageLocations = slices.Clone(rg.offsetIndex[i].PageLocations)
	}

	if reuseRowGroup == nil {
		w.rowGroups = append(w.rowGroups, format.RowGroup{
			Columns:             columns,
			TotalByteSize:       totalByteSize,
			NumRows:             numRows,
			SortingColumns:      sortingColumns,
			FileOffset:          fileOffset,
			TotalCompressedSize: totalCompressedSize,
			Ordinal:             int16(len(w.rowGroups)),
		})
	} else {
		reuseRowGroup.Columns = columns
		reuseRowGroup.TotalByteSize = totalByteSize
		reuseRowGroup.NumRows = numRows
		reuseRowGroup.SortingColumns = sortingColumns
		reuseRowGroup.FileOffset = fileOffset
		reuseRowGroup.TotalCompressedSize = totalCompressedSize
		reuseRowGroup.Ordinal = int16(len(w.rowGroups) - 1)
	}

	return numRows, nil
}

func (w *writer) WriteRows(rows []Row) (written int, err error) {
	var n int
	for len(rows) > 0 {
		n, err = w.currentRowGroup.WriteRows(rows)
		rows = rows[n:]
		written += n
		if err != ErrTooManyRowGroups {
			break
		}
		if err = w.flush(); err != nil {
			break
		}
	}
	return
}

// The WriteValues method is intended to work in pair with WritePage to allow
// programs to target writing values to specific columns of of the writer.
func (w *writer) WriteValues(values []Value) (numValues int, err error) {
	return w.currentRowGroup.columns[values[0].Column()].writeValues(values)
}

var (
	dataPageBuffers       memory.Pool[writerBuffers]
	dictionaryPageBuffers memory.Pool[writerBuffers]
)

// One writerBuffers is used by each writer instance, the memory buffers here
// are shared by all columns of the writer because serialization is not done
// concurrently, which helps keep memory utilization low, both in the total
// footprint and GC cost.
//
// The type also exposes helper methods to facilitate the generation of parquet
// pages. A scratch space is used when serialization requires combining multiple
// buffers or compressing the page data, with double-buffering technique being
// employed by swapping the scratch and page buffers to minimize memory copies.
type writerBuffers struct {
	repetitions []byte // buffer used to encode repetition levels
	definitions []byte // buffer used to encode definition levels
	page        []byte // page buffer holding the page data
	scratch     []byte // scratch space used for compression
}

func newWriterBuffers() *writerBuffers { return new(writerBuffers) }

func (wb *writerBuffers) crc32() (checksum uint32) {
	checksum = crc32.Update(checksum, crc32.IEEETable, wb.repetitions)
	checksum = crc32.Update(checksum, crc32.IEEETable, wb.definitions)
	checksum = crc32.Update(checksum, crc32.IEEETable, wb.page)
	return checksum
}

func (wb *writerBuffers) size() int {
	return len(wb.repetitions) + len(wb.definitions) + len(wb.page)
}

func (wb *writerBuffers) reset() {
	wb.repetitions = wb.repetitions[:0]
	wb.definitions = wb.definitions[:0]
	wb.page = wb.page[:0]
}

func encodeLevels(dst, src []byte, maxLevel byte) ([]byte, error) {
	bitWidth := bits.Len8(maxLevel)
	return levelEncodingsRLE[bitWidth-1].EncodeLevels(dst, src)
}

func (wb *writerBuffers) encodeRepetitionLevels(page Page, maxRepetitionLevel byte) (err error) {
	wb.repetitions, err = encodeLevels(wb.repetitions, page.RepetitionLevels(), maxRepetitionLevel)
	return
}

func (wb *writerBuffers) encodeDefinitionLevels(page Page, maxDefinitionLevel byte) (err error) {
	wb.definitions, err = encodeLevels(wb.definitions, page.DefinitionLevels(), maxDefinitionLevel)
	return
}

func (wb *writerBuffers) prependLevelsToDataPageV1(maxRepetitionLevel, maxDefinitionLevel byte) {
	hasRepetitionLevels := maxRepetitionLevel > 0
	hasDefinitionLevels := maxDefinitionLevel > 0

	if hasRepetitionLevels || hasDefinitionLevels {
		wb.scratch = wb.scratch[:0]
		// In data pages v1, the repetition and definition levels are prefixed
		// with the 4 bytes length of the sections. While the parquet-format
		// documentation indicates that the length prefix is part of the hybrid
		// RLE/Bit-Pack encoding, this is the only condition where it is used
		// so we treat it as a special case rather than implementing it in the
		// encoding.
		//
		// Reference https://github.com/apache/parquet-format/blob/master/Encodings.md#run-length-encoding--bit-packing-hybrid-rle--3
		if hasRepetitionLevels {
			wb.scratch = plain.AppendInt32(wb.scratch, int32(len(wb.repetitions)))
			wb.scratch = append(wb.scratch, wb.repetitions...)
			wb.repetitions = wb.repetitions[:0]
		}
		if hasDefinitionLevels {
			wb.scratch = plain.AppendInt32(wb.scratch, int32(len(wb.definitions)))
			wb.scratch = append(wb.scratch, wb.definitions...)
			wb.definitions = wb.definitions[:0]
		}
		wb.scratch = append(wb.scratch, wb.page...)
		wb.swapPageAndScratchBuffers()
	}
}

func (wb *writerBuffers) encode(page Page, enc encoding.Encoding) (err error) {
	pageType := page.Type()
	pageData := page.Data()
	wb.page, err = pageType.Encode(wb.page[:0], pageData, enc)
	return err
}

func (wb *writerBuffers) compress(codec compress.Codec) (err error) {
	wb.scratch, err = codec.Encode(wb.scratch[:0], wb.page)
	wb.swapPageAndScratchBuffers()
	return err
}

func (wb *writerBuffers) swapPageAndScratchBuffers() {
	wb.page, wb.scratch = wb.scratch, wb.page[:0]
}

// ColumnWriter writes values for a single column to underlying medium.
type ColumnWriter struct {
	pool       BufferPool
	pageBuffer io.ReadWriteSeeker
	numPages   int

	columnPath           columnPath
	columnType           Type
	originalType         Type // Original type before any encoding changes
	columnIndex          ColumnIndexer
	columnBuffer         ColumnBuffer
	plainColumnBuffer    ColumnBuffer // Retained plain buffer for fallback after lazy creation
	originalColumnBuffer ColumnBuffer // Original buffer to restore after row group flush
	columnFilter         BloomFilterColumn
	encoding             encoding.Encoding
	originalEncoding     encoding.Encoding // Original encoding before any changes
	compression          compress.Codec
	dictionary           Dictionary

	maxRepetitionLevel byte
	maxDefinitionLevel byte

	header struct {
		buffer   bytes.Buffer
		protocol thrift.CompactProtocol
		encoder  thrift.Encoder
		dict     format.PageHeader
		page     format.PageHeader
	}

	filter                    []byte
	numRows                   int64
	bufferIndex               int32
	bufferSize                int32
	writePageStats            bool
	writePageBounds           bool
	writeDeprecatedStatistics bool
	isCompressed              bool
	encodings                 []format.Encoding

	columnChunk        *format.ColumnChunk
	offsetIndex        *format.OffsetIndex
	hasSwitchedToPlain bool  // Tracks if dictionary encoding was switched to PLAIN
	dictionaryMaxBytes int64 // Per-column dictionary size limit

	totalUnencodedByteArrayBytes  int64
	repetitionLevelHistogram      []int64
	definitionLevelHistogram      []int64
	pageRepetitionLevelHistograms []int64
	pageDefinitionLevelHistograms []int64
}

func (c *ColumnWriter) reset() {
	if c.hasSwitchedToPlain {
		c.columnType = c.originalType
		c.encoding = c.originalEncoding
		c.hasSwitchedToPlain = false
	}
	if c.originalColumnBuffer != nil {
		c.columnBuffer = c.originalColumnBuffer
	}
	if c.columnBuffer != nil {
		c.columnBuffer.Reset()
	}
	if c.columnIndex != nil {
		c.columnIndex.Reset()
	}
	if c.dictionary != nil {
		c.dictionary.Reset()
	}
	if c.pageBuffer != nil {
		c.pool.PutBuffer(c.pageBuffer)
		c.pageBuffer = nil
	}
	c.numPages = 0
	// Bloom filters may change in size between row groups, but we retain the
	// buffer to avoid reallocating large memory blocks.
	c.filter = c.filter[:0]
	c.numRows = 0
	// Reset the fields of column chunks that change between row groups,
	// but keep the ones that remain unchanged.
	c.columnChunk.MetaData.NumValues = 0
	c.columnChunk.MetaData.TotalUncompressedSize = 0
	c.columnChunk.MetaData.TotalCompressedSize = 0
	c.columnChunk.MetaData.DataPageOffset = 0
	c.columnChunk.MetaData.DictionaryPageOffset = 0
	c.columnChunk.MetaData.Statistics = format.Statistics{}
	c.columnChunk.MetaData.EncodingStats = c.columnChunk.MetaData.EncodingStats[:0]
	c.columnChunk.MetaData.BloomFilterOffset = 0
	c.offsetIndex.PageLocations = c.offsetIndex.PageLocations[:0]

	c.totalUnencodedByteArrayBytes = 0
	clear(c.repetitionLevelHistogram)
	clear(c.definitionLevelHistogram)
	c.pageRepetitionLevelHistograms = c.pageRepetitionLevelHistograms[:0]
	c.pageDefinitionLevelHistograms = c.pageDefinitionLevelHistograms[:0]
}

func (c *ColumnWriter) totalRowCount() int64 {
	n := c.numRows
	if c.columnBuffer != nil {
		n += int64(c.columnBuffer.Len())
	}
	return n
}

// Flush writes any buffered data to the underlying [io.Writer].
func (c *ColumnWriter) Flush() (err error) {
	if c.columnBuffer == nil {
		return nil
	}
	if c.columnBuffer.Len() > 0 {
		// Check dictionary size limit BEFORE writing the page
		// to decide if we should switch to PLAIN for future pages
		var fallbackToPlain bool
		if c.dictionary != nil && !c.hasSwitchedToPlain {
			currentDictSize := c.dictionary.Size()
			// Check user-configured limit
			if c.dictionaryMaxBytes > 0 && currentDictSize > c.dictionaryMaxBytes {
				fallbackToPlain = true
			}
			// For byte array dictionaries, enforce hard limit of 1GB.
			// The offsets are stored as uint32 (max 4GB), but a dictionary larger
			// than 1GB doesn't make practical sense and wastes memory.
			if !fallbackToPlain && c.columnType.Kind() == ByteArray && currentDictSize > 1024*1024*1024 {
				fallbackToPlain = true
			}
		}

		// Write the current buffered page (still with current encoding)
		defer c.columnBuffer.Reset()
		_, err = c.writeDataPage(c.columnBuffer.Page())
		if err != nil {
			return err
		}

		// After writing the page, convert to PLAIN for future pages if needed
		// This avoids wasteful buffer allocation if this was the last page
		if fallbackToPlain {
			if err := c.fallbackDictionaryToPlain(); err != nil {
				return fmt.Errorf("converting dictionary to plain: %w", err)
			}
		}
	}
	return err
}

func (c *ColumnWriter) flushFilterPages() (err error) {
	if c.columnFilter == nil {
		return nil
	}

	// If there is a dictionary, it contains all the values that we need to
	// write to the filter.
	if dict := c.dictionary; dict != nil {
		// Need to always attempt to resize the filter, as the writer might
		// be reused after resetting which would have reset the length of
		// the filter to 0.
		c.resizeBloomFilter(int64(dict.Len()))
		return c.writePageToFilter(dict.Page())
	}

	// When the filter was already allocated, pages have been written to it as
	// they were seen by the column writer.
	if len(c.filter) > 0 {
		return nil
	}

	// Skip columns with nil pageBuffer (e.g., empty struct groups with no leaf columns)
	if c.pageBuffer == nil {
		return nil
	}

	// When the filter was not allocated, the writer did not know how many
	// values were going to be seen and therefore could not properly size the
	// filter ahead of time. In this case, we read back all the pages that we
	// have encoded and copy their values back to the filter.
	//
	// A prior implementation of the column writer used to create in-memory
	// copies of the pages to avoid this decoding step; however, this unbounded
	// allocation caused memory exhaustion in production applications. CPU being
	// a somewhat more stretchable resource, we prefer spending time on this
	// decoding step than having to trigger incident response when production
	// systems are getting OOM-Killed.
	c.resizeBloomFilter(c.columnChunk.MetaData.NumValues)

	column := &Column{
		// Set all the fields required by the decodeDataPage* methods.
		typ:                c.columnType,
		encoding:           c.encoding,
		compression:        c.compression,
		maxRepetitionLevel: c.maxRepetitionLevel,
		maxDefinitionLevel: c.maxDefinitionLevel,
		index:              int16(c.bufferIndex),
	}

	var pageReader io.Reader = c.pageBuffer
	if offset, err := c.pageBuffer.Seek(0, io.SeekStart); err != nil {
		return err
	} else if offset != 0 {
		return fmt.Errorf("resetting parquet page buffer to the start expected offset zero but got %d", offset)
	}

	if _, ok := pageReader.(*os.File); ok {
		rbuf, pool := getBufioReader(pageReader, 1024)
		defer func() {
			putBufioReader(rbuf, pool)
		}()
		pageReader = rbuf
	}

	pbuf := (*buffer[byte])(nil)
	defer func() {
		if pbuf != nil {
			pbuf.unref()
		}
	}()

	decoder := thrift.NewDecoder(c.header.protocol.NewReader(pageReader))

	for range c.numPages {
		header := new(format.PageHeader)
		if err := decoder.Decode(header); err != nil {
			return err
		}

		if pbuf != nil {
			pbuf.unref()
		}
		pbuf = buffers.get(int(header.CompressedPageSize))
		if _, err := io.ReadFull(pageReader, pbuf.data.Slice()); err != nil {
			return err
		}

		var page Page

		switch header.Type {
		case format.DataPage:
			page, err = column.decodeDataPageV1(DataPageHeaderV1{&header.DataPageHeader.V}, pbuf, nil, header.UncompressedPageSize)
		case format.DataPageV2:
			page, err = column.decodeDataPageV2(DataPageHeaderV2{&header.DataPageHeaderV2.V}, pbuf, nil, header.UncompressedPageSize)
		}
		if page != nil {
			err = c.writePageToFilter(page)
			Release(page)
		}
		if err != nil {
			return err
		}
	}

	return nil
}

func (c *ColumnWriter) resizeBloomFilter(numValues int64) {
	filterSize := c.columnFilter.Size(numValues)
	if cap(c.filter) < filterSize {
		c.filter = make([]byte, filterSize)
	} else {
		c.filter = c.filter[:filterSize]
		for i := range c.filter {
			c.filter[i] = 0
		}
	}
}

func (c *ColumnWriter) newColumnBuffer() ColumnBuffer {
	columnIndex := int(c.bufferIndex)
	switch {
	case c.maxRepetitionLevel > 0:
		column := c.columnType.NewColumnBuffer(columnIndex, 0)
		return newRepeatedColumnBuffer(column, c.maxRepetitionLevel, c.maxDefinitionLevel, nullsGoLast)
	case c.maxDefinitionLevel > 0:
		column := c.columnType.NewColumnBuffer(columnIndex, 0)
		return newOptionalColumnBuffer(column, c.maxDefinitionLevel, nullsGoLast)
	default:
		numValues := c.columnType.EstimateNumValues(int(c.bufferSize))
		return c.columnType.NewColumnBuffer(columnIndex, numValues)
	}
}

// WriteRowValues writes entire rows to the column. On success, this returns the
// number of rows written (not the number of values).
//
// Unlike ValueWriter, where arbitrary values may be written regardless of row
// boundaries, this method requires whole rows. This is because the written
// values may be automatically flushed to a data page, based on the writer's
// configured page buffer size, and a single row is not permitted to span two
// pages.
func (c *ColumnWriter) WriteRowValues(rows []Value) (int, error) {
	var startingRows int64
	if c.columnBuffer == nil {
		// Lazily create the row group column so we don't need to allocate it if
		// rows are not written individually to the column.
		c.columnBuffer = c.newColumnBuffer()
		c.originalColumnBuffer = c.columnBuffer
	} else {
		startingRows = int64(c.columnBuffer.Len())
	}
	if _, err := c.columnBuffer.WriteValues(rows); err != nil {
		return 0, err
	}
	numRows := int(int64(c.columnBuffer.Len()) - startingRows)
	if c.columnBuffer.Size() >= int64(c.bufferSize) {
		return numRows, c.Flush()
	}
	return numRows, nil
}

// Close closes the column writer and resets all dependent resources.
// It can be reused after Close is called.
func (c *ColumnWriter) Close() (err error) {
	if c.columnBuffer == nil {
		return nil
	}
	if err := c.Flush(); err != nil {
		return err
	}
	c.columnBuffer.Reset()
	return nil
}

func (c *ColumnWriter) writeValues(values []Value) (numValues int, err error) {
	if c.columnBuffer == nil {
		c.columnBuffer = c.newColumnBuffer()
		// Save the original dictionary-encoding buffer to restore after row group flush
		if c.originalColumnBuffer == nil {
			c.originalColumnBuffer = c.columnBuffer
		}
	}
	return c.columnBuffer.WriteValues(values)
}

func (c *ColumnWriter) writeBloomFilter(w io.Writer) error {
	e := thrift.NewEncoder(c.header.protocol.NewWriter(w))
	h := bloomFilterHeader(c.columnFilter)
	h.NumBytes = int32(len(c.filter))
	if err := e.Encode(&h); err != nil {
		return err
	}
	_, err := w.Write(c.filter)
	return err
}

func (c *ColumnWriter) writeDataPage(page Page) (int64, error) {
	numValues := page.NumValues()
	if numValues == 0 {
		return 0, nil
	}

	buf := dataPageBuffers.Get(newWriterBuffers, (*writerBuffers).reset)
	defer dataPageBuffers.Put(buf)

	if c.maxRepetitionLevel > 0 {
		buf.encodeRepetitionLevels(page, c.maxRepetitionLevel)
	}
	if c.maxDefinitionLevel > 0 {
		buf.encodeDefinitionLevels(page, c.maxDefinitionLevel)
	}

	if err := buf.encode(page, c.encoding); err != nil {
		return 0, fmt.Errorf("encoding parquet data page: %w", err)
	}

	if c.header.page.Type == format.DataPage {
		buf.prependLevelsToDataPageV1(c.maxRepetitionLevel, c.maxDefinitionLevel)
	}

	uncompressedPageSize := buf.size()
	if uncompressedPageSize > maxUncompressedPageSize {
		return 0, fmt.Errorf("page size limit exceeded: %d>%d", uncompressedPageSize, maxUncompressedPageSize)
	}
	if c.isCompressed {
		if err := buf.compress(c.compression); err != nil {
			return 0, fmt.Errorf("compressing parquet data page: %w", err)
		}
	}

	if page.Dictionary() == nil && len(c.filter) > 0 {
		// When the writer knows the number of values in advance (e.g. when
		// writing a full row group), the filter encoding is set and the page
		// can be directly applied to the filter, which minimizes memory usage
		// since there is no need to buffer the values in order to determine
		// the size of the filter.
		if err := c.writePageToFilter(page); err != nil {
			return 0, err
		}
	}

	statistics := format.Statistics{}
	if c.writePageStats {
		statistics = c.makePageStatistics(page)
	}

	c.header.page.UncompressedPageSize = int32(uncompressedPageSize)
	c.header.page.CompressedPageSize = int32(buf.size())
	c.header.page.CRC = int32(buf.crc32())

	numRows := page.NumRows()
	numNulls := page.NumNulls()
	switch {
	case c.header.page.DataPageHeader.Valid:
		c.header.page.DataPageHeader.V = format.DataPageHeader{
			NumValues:               int32(numValues),
			Encoding:                c.encoding.Encoding(),
			DefinitionLevelEncoding: format.RLE,
			RepetitionLevelEncoding: format.RLE,
			Statistics:              statistics,
		}
	case c.header.page.DataPageHeaderV2.Valid:
		c.header.page.DataPageHeaderV2.V = format.DataPageHeaderV2{
			NumValues:                  int32(numValues),
			NumNulls:                   int32(numNulls),
			NumRows:                    int32(numRows),
			Encoding:                   c.encoding.Encoding(),
			DefinitionLevelsByteLength: int32(len(buf.definitions)),
			RepetitionLevelsByteLength: int32(len(buf.repetitions)),
			IsCompressed:               thrift.New(c.isCompressed),
			Statistics:                 statistics,
		}
	}

	c.header.buffer.Reset()
	c.header.buffer.Grow(1024)

	if err := c.header.encoder.Encode(&c.header.page); err != nil {
		return 0, err
	}

	size := int64(c.header.buffer.Len()) +
		int64(len(buf.repetitions)) +
		int64(len(buf.definitions)) +
		int64(len(buf.page))

	err := c.writePageTo(size, func(output io.Writer) (written int64, err error) {
		for _, data := range [...][]byte{
			c.header.buffer.Bytes(),
			buf.repetitions,
			buf.definitions,
			buf.page,
		} {
			wn, err := output.Write(data)
			written += int64(wn)
			if err != nil {
				return written, err
			}
		}
		return written, nil
	})
	if err != nil {
		return 0, err
	}

	c.recordPageStats(int32(c.header.buffer.Len()), &c.header.page, page)
	return numValues, nil
}

func (c *ColumnWriter) writeDictionaryPage(output io.Writer, dict Dictionary) (err error) {
	buf := dictionaryPageBuffers.Get(newWriterBuffers, (*writerBuffers).reset)
	defer dictionaryPageBuffers.Put(buf)

	if err := buf.encode(dict.Page(), &Plain); err != nil {
		return fmt.Errorf("writing parquet dictionary page: %w", err)
	}

	uncompressedPageSize := buf.size()
	if uncompressedPageSize > maxUncompressedPageSize {
		return fmt.Errorf("page size limit exceeded: %d>%d", uncompressedPageSize, maxUncompressedPageSize)
	}
	if isCompressed(c.compression) {
		if err := buf.compress(c.compression); err != nil {
			return fmt.Errorf("compressing parquet dictionary page: %w", err)
		}
	}

	c.header.dict.UncompressedPageSize = int32(uncompressedPageSize)
	c.header.dict.CompressedPageSize = int32(buf.size())
	c.header.dict.CRC = int32(buf.crc32())

	c.header.dict.DictionaryPageHeader.V = format.DictionaryPageHeader{
		NumValues: int32(dict.Len()),
		Encoding:  format.Plain,
		IsSorted:  false,
	}

	c.header.buffer.Reset()
	c.header.buffer.Grow(1024)

	if err := c.header.encoder.Encode(&c.header.dict); err != nil {
		return err
	}
	if _, err := output.Write(c.header.buffer.Bytes()); err != nil {
		return err
	}
	if _, err := output.Write(buf.page); err != nil {
		return err
	}
	c.recordPageStats(int32(c.header.buffer.Len()), &c.header.dict, nil)
	return nil
}

func (c *ColumnWriter) writePageToFilter(page Page) (err error) {
	pageType := page.Type()
	pageData := page.Data()
	c.filter, err = pageType.Encode(c.filter, pageData, c.columnFilter.Encoding())
	return err
}

func (c *ColumnWriter) writePageTo(size int64, writeTo func(io.Writer) (int64, error)) (err error) {
	if c.pageBuffer == nil {
		c.pageBuffer = c.pool.GetBuffer()
		defer func() {
			if err != nil {
				c.pool.PutBuffer(c.pageBuffer)
				c.pageBuffer = nil
			}
		}()
		if _, err = c.pageBuffer.Seek(0, io.SeekStart); err != nil {
			return err
		}
	}
	written, err := writeTo(c.pageBuffer)
	if err != nil {
		return err
	}
	if written != size {
		return fmt.Errorf("writing parquet column page expected %dB but got %dB: %w", size, written, io.ErrShortWrite)
	}
	c.numPages++
	return nil
}

// fallbackDictionaryToPlain switches future pages from dictionary to PLAIN encoding.
// This is called when a column's dictionary size limit is exceeded.
func (c *ColumnWriter) fallbackDictionaryToPlain() error {
	// Switch to PLAIN encoding for future writes
	// Get the underlying type without the indexed wrapper
	if indexedType, ok := c.columnType.(*indexedType); ok {
		c.columnType = indexedType.Type
	}
	if c.plainColumnBuffer == nil {
		base := c.columnType.NewColumnBuffer(int(c.bufferIndex), int(c.bufferSize))
		switch {
		case c.maxRepetitionLevel > 0:
			c.plainColumnBuffer = newRepeatedColumnBuffer(base, c.maxRepetitionLevel, c.maxDefinitionLevel, nullsGoLast)
		case c.maxDefinitionLevel > 0:
			c.plainColumnBuffer = newOptionalColumnBuffer(base, c.maxDefinitionLevel, nullsGoLast)
		default:
			c.plainColumnBuffer = base
		}
	}
	c.columnBuffer = c.plainColumnBuffer
	c.encoding = &plain.Encoding{}
	c.encodings = addEncoding(c.encodings, format.Plain)
	// DON'T clear the dictionary reference!
	// We need to keep it so:
	// 1. The dictionary page can be written (required for existing dict-encoded pages)
	// 2. Existing pages that were written with dictionary indexes can be read
	//
	// The hasSwitchedToPlain flag prevents new values from being added to the dictionary
	//
	// Note: We are NOT re-encoding existing pages. This means:
	// - Pages written before the limit was hit will remain dictionary-encoded
	// - Pages written after will be PLAIN-encoded
	// - The dictionary page will still be written at row group flush
	// - Mixed encodings in the same column chunk are valid per Parquet spec
	c.hasSwitchedToPlain = true
	return nil
}

func (c *ColumnWriter) makePageStatistics(page Page) format.Statistics {
	numNulls := page.NumNulls()
	minValue, maxValue, _ := page.Bounds()
	minValueBytes := minValue.Bytes()
	maxValueBytes := maxValue.Bytes()
	return format.Statistics{
		Min:       minValueBytes, // deprecated
		Max:       maxValueBytes, // deprecated
		NullCount: numNulls,
		MinValue:  minValueBytes,
		MaxValue:  maxValueBytes,
	}
}

func (c *ColumnWriter) recordPageStats(headerSize int32, header *format.PageHeader, page Page) {
	uncompressedSize := headerSize + header.UncompressedPageSize
	compressedSize := headerSize + header.CompressedPageSize

	if page != nil {
		numNulls := page.NumNulls()
		numValues := page.NumValues()

		var minValue, maxValue Value
		var pageHasBounds bool
		if c.writePageBounds {
			minValue, maxValue, pageHasBounds = page.Bounds()
		}

		c.columnIndex.IndexPage(numValues, numNulls, minValue, maxValue)
		c.columnChunk.MetaData.NumValues += numValues
		c.columnChunk.MetaData.Statistics.NullCount += numNulls

		if pageHasBounds {
			var existingMaxValue, existingMinValue Value

			if c.columnChunk.MetaData.Statistics.MaxValue != nil && c.columnChunk.MetaData.Statistics.MinValue != nil {
				existingMaxValue = c.columnType.Kind().Value(c.columnChunk.MetaData.Statistics.MaxValue)
				existingMinValue = c.columnType.Kind().Value(c.columnChunk.MetaData.Statistics.MinValue)
			}

			if existingMaxValue.isNull() || c.columnType.Compare(maxValue, existingMaxValue) > 0 {
				buf := c.columnChunk.MetaData.Statistics.MaxValue[:0]
				// if maxValue is empty string, c.columnChunk.MetaData.Statistics.MaxValue should be []bytes{}, but nil
				if buf == nil && maxValue.Kind() == ByteArray && len(maxValue.ByteArray()) == 0 {
					buf = make([]byte, 0)
				}
				c.columnChunk.MetaData.Statistics.MaxValue = maxValue.AppendBytes(buf)
				if c.writeDeprecatedStatistics {
					c.columnChunk.MetaData.Statistics.Max = c.columnChunk.MetaData.Statistics.MaxValue
				}
			}

			if existingMinValue.isNull() || c.columnType.Compare(minValue, existingMinValue) < 0 {
				buf := c.columnChunk.MetaData.Statistics.MinValue[:0]
				// same as above
				if buf == nil && minValue.Kind() == ByteArray && len(minValue.ByteArray()) == 0 {
					buf = make([]byte, 0)
				}
				c.columnChunk.MetaData.Statistics.MinValue = minValue.AppendBytes(buf)
				if c.writeDeprecatedStatistics {
					c.columnChunk.MetaData.Statistics.Min = c.columnChunk.MetaData.Statistics.MinValue
				}
			}
		}

		c.offsetIndex.PageLocations = append(c.offsetIndex.PageLocations, format.PageLocation{
			Offset:             c.columnChunk.MetaData.TotalCompressedSize,
			CompressedPageSize: compressedSize,
			FirstRowIndex:      c.numRows,
		})

		c.numRows += page.NumRows()
		c.totalUnencodedByteArrayBytes += computeUnencodedByteArraySize(page)

		repetitionLevels := page.RepetitionLevels()
		definitionLevels := page.DefinitionLevels()

		if c.maxRepetitionLevel > 0 {
			c.pageRepetitionLevelHistograms = accumulateAndAppendPageLevelHistogram(
				c.repetitionLevelHistogram,
				c.pageRepetitionLevelHistograms,
				repetitionLevels,
				c.maxRepetitionLevel,
			)
		}

		if c.maxDefinitionLevel > 0 {
			c.pageDefinitionLevelHistograms = accumulateAndAppendPageLevelHistogram(
				c.definitionLevelHistogram,
				c.pageDefinitionLevelHistograms,
				definitionLevels,
				c.maxDefinitionLevel,
			)
		}
	}

	pageType := header.Type
	encoding := format.Encoding(-1)
	switch pageType {
	case format.DataPageV2:
		encoding = header.DataPageHeaderV2.V.Encoding
	case format.DataPage:
		encoding = header.DataPageHeader.V.Encoding
	case format.DictionaryPage:
		encoding = header.DictionaryPageHeader.V.Encoding
	}

	c.columnChunk.MetaData.TotalUncompressedSize += int64(uncompressedSize)
	c.columnChunk.MetaData.TotalCompressedSize += int64(compressedSize)
	c.columnChunk.MetaData.EncodingStats = addPageEncodingStats(c.columnChunk.MetaData.EncodingStats, format.PageEncodingStats{
		PageType: pageType,
		Encoding: encoding,
		Count:    1,
	})
}

func addEncoding(encodings []format.Encoding, add format.Encoding) []format.Encoding {
	if slices.Contains(encodings, add) {
		return encodings
	}
	return append(encodings, add)
}

func addPageEncodingStats(stats []format.PageEncodingStats, pages ...format.PageEncodingStats) []format.PageEncodingStats {
addPages:
	for _, add := range pages {
		for i, st := range stats {
			if st.PageType == add.PageType && st.Encoding == add.Encoding {
				stats[i].Count += add.Count
				continue addPages
			}
		}
		stats = append(stats, add)
	}
	return stats
}

func sortPageEncodings(encodings []format.Encoding) {
	slices.Sort(encodings)
}

func sortPageEncodingStats(stats []format.PageEncodingStats) {
	slices.SortFunc(stats, func(s1, s2 format.PageEncodingStats) int {
		if k := cmp.Compare(s1.PageType, s2.PageType); k != 0 {
			return k
		}
		return cmp.Compare(s1.Encoding, s2.Encoding)
	})
}

type offsetTrackingWriter struct {
	writer io.Writer
	offset int64
}

func (w *offsetTrackingWriter) Reset(writer io.Writer) {
	w.writer = writer
	w.offset = 0
}

func (w *offsetTrackingWriter) Write(b []byte) (int, error) {
	n, err := w.writer.Write(b)
	w.offset += int64(n)
	return n, err
}

func (w *offsetTrackingWriter) WriteString(s string) (int, error) {
	n, err := io.WriteString(w.writer, s)
	w.offset += int64(n)
	return n, err
}

func (w *offsetTrackingWriter) ReadFrom(r io.Reader) (int64, error) {
	// io.Copy will make use of io.ReaderFrom if w.writer implements it.
	n, err := io.Copy(w.writer, r)
	w.offset += n
	return n, err
}

var (
	_ RowWriterWithSchema = (*Writer)(nil)
	_ RowReaderFrom       = (*Writer)(nil)
	_ RowGroupWriter      = (*Writer)(nil)

	_ RowWriter   = (*writer)(nil)
	_ ValueWriter = (*writer)(nil)

	_ io.ReaderFrom   = (*offsetTrackingWriter)(nil)
	_ io.StringWriter = (*offsetTrackingWriter)(nil)
)
