// Cross-boundary fusion for [aggregating boundary] -> [identity ORDER BY
// boundary] -> [DISTINCT column-subset + LIMIT]: DISTINCT keeps each
// distinct tuple's FIRST row under the sort's total order and LIMIT keeps
// the leading bound, so each tuple's effective position is the MINIMUM of
// its rows' key vectors (argmin) and the result is a bounded selection
// over those minima. Streaming the aggregate's finalized groups through a
// per-tuple argmin map replaces the full group-row materialization, the
// full-width sort decoration, and the sort itself.
package exec

import (
	
	

	
	
	
	
)

// orderedDistinctFusions counts fused chain executions (the engagement
// oracle); disableOrderedDistinct pins differential tests to the general
// materialize-sort-dedup pipeline.
var (
	orderedDistinctFusions int
	disableOrderedDistinct bool
)

// orderedDistinctShape validates the three-segment shape and resolves the
// DISTINCT boundary's column reads. The ordering boundary must be a pure
// identity carrying only ORDER BY (no pagination, no filter), and the
// DISTINCT boundary must be a bare column-subset projection with a LIMIT
// and nothing else -- anything richer falls back to the general path.
func orderedDistinctShape( []*plan.Segment,  int) ( []int,  int,  bool) {
	if +2 >= len() {
		return nil, 0, false
	}
	, ,  := [], [+1], [+2]
	if !.Proj.Aggregated || .Proj.Distinct || len(.Proj.OrderBy) != 0 ||
		.Proj.Skip != nil || .Proj.Limit != nil || .PostWhere != nil {
		return nil, 0, false
	}
	if !identityBoundary() || len(.Proj.OrderBy) == 0 || .Proj.Skip != nil ||
		.Proj.Limit != nil || .PostWhere != nil || .RowWidth != len(.Proj.Returns) {
		return nil, 0, false
	}
	if len(.Stages) != 0 || !.Proj.Distinct || .Proj.Aggregated ||
		len(.Proj.OrderBy) != 0 || .Proj.Limit == nil || .PostWhere != nil ||
		len(.Proj.Post) != 0 || .Proj.NHidden != 0 {
		return nil, 0, false
	}
	 = make([]int, len(.Proj.Returns))
	for ,  := range .Proj.Returns {
		,  := .Expr.(*ast.Var)
		if ! {
			return nil, 0, false
		}
		,  := .Slots[.Name]
		if ! ||  < 0 ||  >= .RowWidth {
			return nil, 0, false
		}
		[] = 
	}
	 := orderBound(&.Proj)
	if  < 0 {
		return nil, 0, false
	}
	return , , true
}

// tryOrderedDistinctTopK recognizes the shape at segs[at], executes the
// aggregate segment normally, streams its finalized groups through a
// distinct-tuple argmin, bounded-selects the winning tuples, and returns
// the DISTINCT boundary's output. consumed is 3 on success.
func tryOrderedDistinctTopK( *eval.Ctx,  []*plan.Segment,  int,  [][]value.Value) ([][]value.Value, int, bool) {
	if disableOrderedDistinct {
		return nil, 0, false
	}
	, ,  := orderedDistinctShape(, )
	if ! {
		return nil, 0, false
	}
	, ,  := [], [+1], [+2]
	orderedDistinctFusions++

	// Run the aggregate segment's stage chain into its sink, exactly as
	// runSegmentRun would for a run head.
	 := newAggSink(, &.Proj, .Slots)
	 := segmentBoundSlots()
	var  []value.Value
	if len() > 0 {
		 = make([]value.Value, .RowWidth)
		copy(, [0])
	}
	 := func( int) bool {
		if  < 0 ||  >= len() || [] {
			return false
		}
		return slotAgrees(, , true)
	}
	 := func( int) bool {
		return  >= 0 && len() > 0 && slotAgrees(, , true)
	}
	 := buildChain(, , , , , )
	 := make([]value.Value, .RowWidth)
	for ,  := range  {
		clear()
		copy(, )
		if !.push() {
			break
		}
	}
	.close()

	// Order-key evaluation state, mirroring sortRowsByOrder's scope for
	// the identity boundary (columns at base 0; non-column keys evaluate
	// over the finalized group row).
	 := &.Proj
	 := len(.OrderBy)
	 := make([]int, )
	for  := range .OrderBy {
		[] = plan.OrderColIndex(.OrderBy[].Expr, .Columns, .Returns)
	}
	 := make(map[string]int, len(.Slots)+len(.Columns))
	maps.Copy(, .Slots)
	for ,  := range .Columns {
		[] = 
	}
	 := make([]bool, )
	for  := range .OrderBy {
		[] = .OrderBy[].Desc
	}

	// Bounded argmin: a tuple only matters if its minimum key vector can
	// enter the top bound, per-tuple minima only improve, and the bound-th
	// candidate threshold only tightens -- so a row whose key cannot beat
	// the threshold for an UNTRACKED tuple can never matter (any later,
	// better row of that tuple is judged on its own key), and an evicted
	// candidate's stale minimum is irrelevant (it was already worse than
	// the threshold a re-entry would have to beat). State is O(bound):
	// fixed slot arrays, a worst-at-root heap over slots, and a membership
	// map holding only current candidates.
	 := newArgminTopK(, , len(), )
	 := make([]value.Value, )
	 := 0
	.agg.forEachGroup(, &.Proj, func( []value.Value) {
		for  := range .OrderBy {
			if  := [];  >= 0 {
				[] = []
			} else {
				[] = eval.Eval(, .OrderBy[].Expr, , )
			}
		}
		.offer(, , , )
		++
	})
	return paginate(.sorted(), .Proj.Skip, .Proj.Limit), 3, true
}

// argminTopK is the bounded per-tuple argmin accumulator: at most bound
// distinct tuples are tracked, each with its best (minimum) key vector
// and that row's sequence under the sort's total order. The comparator
// must be EXACTLY the total order the unbounded path sorts by -- here
// (keys, then group arrival sequence, sortRowsByOrder's index tiebreak);
// an engine whose sort breaks ties differently must swap the comparator
// or bounded selection admits a different set than sort-then-truncate.
type argminTopK struct {
	bound, nk, width int
	desc             []bool
	tuples           []value.Value // slot*width flat
	keys             []value.Value // slot*nk flat
	seqs             []int
	heap             []int // heap of slots, worst candidate at root
	pos              []int // slot -> heap index
	n                int
	byPacked         map[uint64]int // packed tuple identity -> slot
	byBytes          map[string]int
	slotPacked       []bool // per-slot identity for eviction
	slotPK           []uint64
	slotSK           []string
	dkey             []byte
}

func newArgminTopK(, ,  int,  []bool) *argminTopK {
	return &argminTopK{
		bound: , nk: , width: , desc: ,
		tuples:   make([]value.Value, *),
		keys:     make([]value.Value, *),
		seqs:     make([]int, ),
		heap:     make([]int, 0, ),
		pos:      make([]int, ),
		byPacked: map[uint64]int{}, byBytes: map[string]int{},
		slotPacked: make([]bool, ),
		slotPK:     make([]uint64, ),
		slotSK:     make([]string, ),
	}
}

// less reports whether (kbuf, seq) sorts strictly before slot's entry.
func ( *argminTopK) ( []value.Value, ,  int) bool {
	for  := range .nk {
		 := value.OrderCmp([], .keys[*.nk+])
		if .desc[] {
			 = -
		}
		if  != 0 {
			return  < 0
		}
	}
	return  < .seqs[]
}

// cmpSlots orders two slots under the same total order.
func ( *argminTopK) (,  int) int {
	if .less(.keys[*.nk:(+1)*.nk], .seqs[], ) {
		return -1
	}
	return 1
}

// offer routes one group row: an existing candidate improves in place, a
// new tuple enters when under capacity or beating the worst candidate
// (which is then evicted, membership and all), anything else is skipped.
func ( *argminTopK) ( []value.Value,  []int,  []value.Value,  int) {
	if .bound == 0 {
		return
	}
	 := false
	var  uint64
	if len() == 1 {
		,  = packGroupKey1([[0]])
	}
	if ! {
		.dkey = .dkey[:0]
		for ,  := range  {
			.dkey = value.AppendKey(.dkey, [])
		}
	}
	,  := -1, false
	if  {
		,  = .byPacked[]
	} else {
		,  = .byBytes[string(.dkey)]
	}
	if  {
		if .less(, , ) {
			copy(.keys[*.nk:(+1)*.nk], )
			.seqs[] = 
			.siftDown(.pos[])
		}
		return
	}
	if .n < .bound {
		 = .n
		.n++
		.fillSlot(, , , , , , )
		.heap = append(.heap, )
		.pos[] = len(.heap) - 1
		.siftUp(len(.heap) - 1)
		return
	}
	 := .heap[0]
	if !.less(, , ) {
		return
	}
	if .slotPacked[] {
		delete(.byPacked, .slotPK[])
	} else {
		delete(.byBytes, .slotSK[])
	}
	.fillSlot(, , , , , , )
	.siftDown(0)
}

func ( *argminTopK) ( int,  []value.Value,  []int,  []value.Value,  int,  bool,  uint64) {
	for ,  := range  {
		.tuples[*.width+] = []
	}
	copy(.keys[*.nk:(+1)*.nk], )
	.seqs[] = 
	.slotPacked[] = 
	if  {
		.slotPK[] = 
		.byPacked[] = 
	} else {
		 := string(.dkey)
		.slotSK[] = 
		.byBytes[] = 
	}
}

func ( *argminTopK) (,  int) {
	.heap[], .heap[] = .heap[], .heap[]
	.pos[.heap[]], .pos[.heap[]] = , 
}

func ( *argminTopK) ( int) {
	for  > 0 {
		 := ( - 1) / 2
		if .cmpSlots(.heap[], .heap[]) <= 0 {
			return
		}
		.swap(, )
		 = 
	}
}

func ( *argminTopK) ( int) {
	for {
		 := 2* + 1
		if  >= len(.heap) {
			return
		}
		 := 
		if  :=  + 1;  < len(.heap) && .cmpSlots(.heap[], .heap[]) > 0 {
			 = 
		}
		if .cmpSlots(.heap[], .heap[]) <= 0 {
			return
		}
		.swap(, )
		 = 
	}
}

// sorted returns the surviving tuples in final order.
func ( *argminTopK) () [][]value.Value {
	 := make([]int, .n)
	for  := range  {
		[] = 
	}
	slices.SortFunc(, .cmpSlots)
	 := make([][]value.Value, .n)
	for ,  := range  {
		[] = .tuples[*.width : *.width+.width : *.width+.width]
	}
	return 
}