// Licensed to the LF AI & Data foundation 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 storage

import (
	"container/heap"
	"io"
	"slices"
	"time"

	"github.com/apache/arrow/go/v17/arrow"
	"github.com/apache/arrow/go/v17/arrow/array"

	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
)

// SortTimings holds phase-level timing information from the Sort function.
type SortTimings struct {
	ReadCost   time.Duration
	SortCost   time.Duration
	WriteCost  time.Duration
	NumBatches int
	NumRows    int
}

// Sort materializes the records from rr, stable-selects the rows for which
// predicate returns true, sorts them by sortByFieldIDs, and writes them out
// through rw in batches of roughly batchSize bytes.
//
// Performance notes (vs. the naive row-at-a-time approach):
//   - The row selection is kept in a value slice ([]rowIndex) instead of a
//     []*rowIndex, avoiding one heap allocation per row.
//   - Sort keys are extracted into flat per-record slices once. A single int64
//     key (the common PK case) is then sorted with an O(N) stable LSD radix
//     sort; other keys use slices.SortFunc over the flat keys (plain slice
//     indexing, no Column() map lookup per comparison).
//   - When writing the output, each source column's array is resolved once per
//     input record rather than once per row (RecordBuilder.Append would do the
//     latter); rows are then emitted in order and flushed once the accumulated
//     batch reaches batchSize bytes.
func Sort(batchSize uint64, schema *schemapb.CollectionSchema, rr []RecordReader,
	rw RecordWriter, predicate func(r Record, ri, i int) bool, sortByFieldIDs []int64,
) (int, *SortTimings, error) {
	records := make([]Record, 0)
	indices := make([]rowIndex, 0)

	// release cgo records
	defer func() {
		for _, rec := range records {
			rec.Release()
		}
	}()

	phaseStart := time.Now()
	for _, r := range rr {
		for {
			rec, err := r.Next()
			if err == nil {
				rec.Retain()
				ri := len(records)
				records = append(records, rec)
				for i := 0; i < rec.Len(); i++ {
					if predicate(rec, ri, i) {
						indices = append(indices, rowIndex{int32(ri), int32(i)})
					}
				}
			} else if err == io.EOF {
				break
			} else {
				return 0, nil, err
			}
		}
	}
	readCost := time.Since(phaseStart)

	if len(records) == 0 {
		return 0, &SortTimings{ReadCost: readCost}, nil
	}

	phaseStart = time.Now()
	if len(sortByFieldIDs) > 0 {
		// Pre-extract the sort key columns into flat per-record slices so the
		// comparator avoids a Column() map lookup + type assert per comparison.
		const (
			keyInt64 = iota
			keyString
		)
		kinds := make([]int, len(sortByFieldIDs))
		int64Keys := make([][][]int64, len(sortByFieldIDs))
		stringKeys := make([][][]string, len(sortByFieldIDs))
		for fp, fid := range sortByFieldIDs {
			switch records[0].Column(fid).(type) {
			case *array.Int64:
				kinds[fp] = keyInt64
				cols := make([][]int64, len(records))
				for ri, rec := range records {
					cols[ri] = rec.Column(fid).(*array.Int64).Int64Values()
				}
				int64Keys[fp] = cols
			case *array.String:
				kinds[fp] = keyString
				cols := make([][]string, len(records))
				for ri, rec := range records {
					a := rec.Column(fid).(*array.String)
					vals := make([]string, a.Len())
					for i := range vals {
						vals[i] = a.Value(i)
					}
					cols[ri] = vals
				}
				stringKeys[fp] = cols
			default:
				return 0, nil, merr.WrapErrStorageMsg("unsupported type for sorting key")
			}
		}

		// A single int64 sort key (the common PK case) is sorted with a stable
		// LSD radix sort: O(N) instead of O(N log N) and no comparator calls.
		// Multi-field or varchar keys fall back to comparison sort.
		if len(sortByFieldIDs) == 1 && kinds[0] == keyInt64 {
			radixSortByInt64(indices, int64Keys[0])
		} else {
			slices.SortFunc(indices, func(x, y rowIndex) int {
				for fp := range sortByFieldIDs {
					switch kinds[fp] {
					case keyInt64:
						xv, yv := int64Keys[fp][x.ri][x.i], int64Keys[fp][y.ri][y.i]
						if xv != yv {
							if xv < yv {
								return -1
							}
							return 1
						}
					case keyString:
						xv, yv := stringKeys[fp][x.ri][x.i], stringKeys[fp][y.ri][y.i]
						if xv != yv {
							if xv < yv {
								return -1
							}
							return 1
						}
					}
				}
				return 0
			})
		}
	}
	sortCost := time.Since(phaseStart)

	phaseStart = time.Now()
	rb := NewRecordBuilder(schema)

	// Resolve each output column's source array once per input record (instead
	// of once per row, as RecordBuilder.Append would).
	srcByField := make([][]arrow.Array, len(rb.builders))
	defaults := make([]*schemapb.ValueField, len(rb.builders))
	for fi := range rb.builders {
		fid := rb.fields[fi].FieldID
		cols := make([]arrow.Array, len(records))
		for ri := range records {
			cols[ri] = records[ri].Column(fid)
		}
		srcByField[fi] = cols
		defaults[fi] = rb.fields[fi].GetDefaultValue()
	}

	writeRecord := func() error {
		rec := rb.Build()
		defer rec.Release()
		if rec.Len() > 0 {
			return rw.Write(rec)
		}
		return nil
	}

	for _, idx := range indices {
		for fi, builder := range rb.builders {
			size, err := appendValueAt(builder, srcByField[fi][idx.ri], int(idx.i), defaults[fi])
			if err != nil {
				return 0, nil, merr.Wrapf(err, "failed to append value at row %d for field %s", idx.i, rb.fields[fi].GetName())
			}
			rb.size += size
		}
		rb.nRows++

		// Flush once the accumulated batch reaches batchSize bytes (exact, like
		// the original) so a single output record never exceeds the target.
		if rb.GetSize() >= batchSize {
			if err := writeRecord(); err != nil {
				return 0, nil, err
			}
		}
	}

	// write the last partial batch
	if err := writeRecord(); err != nil {
		return 0, nil, err
	}
	writeCost := time.Since(phaseStart)

	timings := &SortTimings{
		ReadCost:   readCost,
		SortCost:   sortCost,
		WriteCost:  writeCost,
		NumBatches: len(records),
		NumRows:    len(indices),
	}
	return len(indices), timings, nil
}

// rowIndex addresses a single row as (record index, row-in-record index). It is
// stored by value to avoid a per-row heap allocation.
type rowIndex struct {
	ri int32
	i  int32
}

// radixSortByInt64 sorts indices in place so that keys[indices[k].ri][indices[k].i]
// is non-decreasing, using a stable LSD radix sort over the 8 bytes of the int64
// key (O(N)). The sign bit is flipped so unsigned byte ordering matches signed
// int64 ordering.
func radixSortByInt64(indices []rowIndex, keys [][]int64) {
	n := len(indices)
	if n < 2 {
		return
	}
	srcKey := make([]uint64, n)
	for i, idx := range indices {
		srcKey[i] = uint64(keys[idx.ri][idx.i]) ^ (uint64(1) << 63)
	}
	dstKey := make([]uint64, n)
	srcIdx := indices
	dstIdx := make([]rowIndex, n)
	var counts [256]int
	for shift := uint(0); shift < 64; shift += 8 {
		counts = [256]int{}
		for i := 0; i < n; i++ {
			counts[(srcKey[i]>>shift)&0xff]++
		}
		sum := 0
		for b := 0; b < 256; b++ {
			c := counts[b]
			counts[b] = sum
			sum += c
		}
		for i := 0; i < n; i++ {
			b := (srcKey[i] >> shift) & 0xff
			p := counts[b]
			counts[b]++
			dstIdx[p] = srcIdx[i]
			dstKey[p] = srcKey[i]
		}
		srcIdx, dstIdx = dstIdx, srcIdx
		srcKey, dstKey = dstKey, srcKey
	}
	// 8 passes is even, so the sorted data ends up back in the original `indices`
	// backing array; copy defensively in case the pass count ever becomes odd.
	if &srcIdx[0] != &indices[0] {
		copy(indices, srcIdx)
	}
}

// A PriorityQueue implements heap.Interface and holds Items.
type PriorityQueue[T any] struct {
	items []*T
	less  func(x, y *T) bool
}

var _ heap.Interface = (*PriorityQueue[any])(nil)

func (pq PriorityQueue[T]) Len() int { return len(pq.items) }

func (pq PriorityQueue[T]) Less(i, j int) bool {
	return pq.less(pq.items[i], pq.items[j])
}

func (pq PriorityQueue[T]) Swap(i, j int) {
	pq.items[i], pq.items[j] = pq.items[j], pq.items[i]
}

func (pq *PriorityQueue[T]) Push(x any) {
	pq.items = append(pq.items, x.(*T))
}

func (pq *PriorityQueue[T]) Pop() any {
	old := pq.items
	n := len(old)
	x := old[n-1]
	old[n-1] = nil
	pq.items = old[0 : n-1]
	return x
}

func (pq *PriorityQueue[T]) Enqueue(x *T) {
	heap.Push(pq, x)
}

func (pq *PriorityQueue[T]) Dequeue() *T {
	return heap.Pop(pq).(*T)
}

func NewPriorityQueue[T any](less func(x, y *T) bool) *PriorityQueue[T] {
	pq := PriorityQueue[T]{
		items: make([]*T, 0),
		less:  less,
	}
	heap.Init(&pq)
	return &pq
}

func MergeSort(batchSize uint64, schema *schemapb.CollectionSchema, rr []RecordReader,
	rw RecordWriter, predicate func(r Record, ri, i int) bool, sortedByFieldIDs []int64,
) (numRows int, err error) {
	// Fast path: no readers provided
	if len(rr) == 0 {
		return 0, nil
	}

	type index struct {
		ri int
		i  int
	}

	recs := make([]Record, len(rr))
	advanceRecord := func(i int) error {
		rec, err := rr[i].Next()
		recs[i] = rec // assign nil if err
		return err
	}

	for i := range rr {
		err := advanceRecord(i)
		if err == io.EOF {
			continue
		}
		if err != nil {
			return 0, err
		}
	}

	comparators := make([]func(x, y *index) int, 0, len(sortedByFieldIDs))
	for _, fid := range sortedByFieldIDs {
		switch recs[0].Column(fid).(type) {
		case *array.Int64:
			comparators = append(comparators, func(x, y *index) int {
				xVal := recs[x.ri].Column(fid).(*array.Int64).Value(x.i)
				yVal := recs[y.ri].Column(fid).(*array.Int64).Value(y.i)
				if xVal < yVal {
					return -1
				}
				if xVal > yVal {
					return 1
				}
				return 0
			})
		case *array.String:
			comparators = append(comparators, func(x, y *index) int {
				xVal := recs[x.ri].Column(fid).(*array.String).Value(x.i)
				yVal := recs[y.ri].Column(fid).(*array.String).Value(y.i)
				if xVal < yVal {
					return -1
				}
				if xVal > yVal {
					return 1
				}
				return 0
			})
		default:
			return 0, merr.WrapErrStorageMsg("unsupported type for sorting key")
		}
	}

	pq := NewPriorityQueue(func(x, y *index) bool {
		for _, cmp := range comparators {
			c := cmp(x, y)
			if c < 0 {
				return true
			}
			if c > 0 {
				return false
			}
		}
		if x.ri != y.ri {
			return x.ri < y.ri
		}
		return x.i < y.i
	})

	endPositions := make([]int, len(recs))
	var enqueueAll func(ri int) error
	enqueueAll = func(ri int) error {
		r := recs[ri]
		hasValid := false
		endPosition := 0
		for j := 0; j < r.Len(); j++ {
			if predicate(r, ri, j) {
				pq.Enqueue(&index{
					ri: ri,
					i:  j,
				})
				numRows++
				hasValid = true
				endPosition = j
			}
		}
		if !hasValid {
			err := advanceRecord(ri)
			if err == io.EOF {
				return nil
			}
			if err != nil {
				return err
			}
			return enqueueAll(ri)
		}
		endPositions[ri] = endPosition
		return nil
	}

	for i, v := range recs {
		if v != nil {
			if err := enqueueAll(i); err != nil {
				return 0, err
			}
		}
	}

	rb := NewRecordBuilder(schema)
	writeRecord := func() error {
		rec := rb.Build()
		defer rec.Release()
		if rec.Len() > 0 {
			return rw.Write(rec)
		}
		return nil
	}

	for pq.Len() > 0 {
		idx := pq.Dequeue()
		if err := rb.Append(recs[idx.ri], idx.i, idx.i+1); err != nil {
			return 0, err
		}
		// Due to current arrow impl (v12), the write performance is largely dependent on the batch size,
		//	small batch size will cause write performance degradation. To work around this issue, we accumulate
		//	records and write them in batches. This requires additional memory copy.
		if rb.GetSize() >= batchSize {
			if err := writeRecord(); err != nil {
				return 0, err
			}
		}

		// If the popped idx reaches the last valid data of the segment, invalidate the cache and advance to the next record
		if idx.i == endPositions[idx.ri] {
			err := advanceRecord(idx.ri)
			if err == io.EOF {
				continue
			}
			if err != nil {
				return 0, err
			}
			if err := enqueueAll(idx.ri); err != nil {
				return 0, err
			}
		}
	}

	// write the last batch
	if rb.GetRowNum() > 0 {
		if err := writeRecord(); err != nil {
			return 0, err
		}
	}

	return numRows, nil
}