// Aggregation execution: routing each matched row into its group's
// accumulators (update), assembling one output row per group with the
// nested-aggregate wrappers, ordering, and pagination (finalize), the
// aggregated terminal sink, and percentile finalization. Split from
// aggregate.go, which holds the aggregator state and group-key packing.
package exec

import (
	
	
	
	
	

	
	
	
)

// aggTopkBuilds counts group rows materialized on the aggregated bounded
// ORDER BY + LIMIT path (the streaming-selection analogue of
// topkPayloadBuilds); disableAggTopk pins differential tests to the
// materialize-everything-then-sort path.
var (
	aggTopkBuilds  int
	disableAggTopk bool
)

// update routes one matched row into its group. Bare-variable key tuples
// probe the packed index straight off the row slots; a hit skips the key
// evaluation and buffering entirely, while a pack failure or miss takes
// the generic path (whose evaluation yields the identical values).
func ( *aggregator) ( *eval.Ctx,  []value.Value,  *plan.ProjPlan,  map[string]int) {
	 := -1
	if .keySlots != nil {
		var  uint64
		var  bool
		if len(.keySlots) == 1 {
			,  = packGroupKey1([.keySlots[0]])
		} else {
			,  = packGroupKey2([.keySlots[0]], [.keySlots[1]])
		}
		if  {
			// One probe covers both outcomes: a miss materializes the key
			// values (identical to their evaluation -- bare slot reads) only
			// to seed the new group.
			 = .indexI.GetOrCreate(, func() int {
				.keyScratch = .keyScratch[:0]
				for ,  := range .keySlots {
					.keyScratch = append(.keyScratch, [])
				}
				return .appendGroup(.keyScratch)
			})
		}
	}
	if  < 0 {
		.keyScratch = .keyScratch[:0]
		for ,  := range .groupC {
			.keyScratch = append(.keyScratch, .Eval(, , ))
		}
		 = .groupIdx(.keyScratch)
	}
	 := .statesOf()
	var  []distinctSet
	if .hasDistinct {
		 = .seenOf()
	}
	var  []value.Value
	if .hasMinMax {
		 = .mmOf()
	}
	var  [][]value.Value
	if .hasCollect {
		 = .itemsOf()
	}
	for  := range .Aggs {
		var  value.Value
		 := .aggC[] != nil
		if  {
			if  := .argSlots[];  >= 0 {
				 = []
			} else {
				 = .aggC[].Eval(, , )
			}
		}
		if .Aggs[].Distinct &&  && !.IsNull() {
			if ![].add(, &.dkScratch) {
				continue
			}
		}
		switch .kinds[] {
		case plan.AggMin, plan.AggMax:
			// The extremum lives on the mm slab; a null slot is the
			// uninitialized sentinel (a min/max arg is never null, so the
			// first non-null arg always seeds it, matching the prior hasMM
			// flag exactly).
			if .IsNull() {
				continue
			}
			if [].IsNull() {
				[] = 
			} else if ,  := value.Compare(, []);  &&
				((.kinds[] == plan.AggMin &&  < 0) || (.kinds[] == plan.AggMax &&  > 0)) {
				[] = 
			}
		case plan.AggCollect:
			if !.IsNull() {
				[] = append([], )
			}
		case plan.AggPercentileCont, plan.AggPercentileDisc:
			// Percentiles are over numbers; non-numeric args skip, like avg.
			if ,  := .AsFloat();  {
				[] = append([], )
			}
		default:
			[].update(, )
		}
	}
}

// postCompile builds the post-aggregation wrapper scope and compiled
// evaluators: wrappers read the hidden accumulator slots as __agg{k} and
// the grouping-key columns by name (the Rust engine's tasks/150); both
// are filled before the wrappers run.
func ( *aggregator) ( *eval.Ctx,  *plan.ProjPlan) (map[string]int, []RowEval) {
	 := len(.Returns)
	 := make(map[string]int, .NHidden+len(.GroupIdx))
	for  := 0;  < .NHidden; ++ {
		[hiddenAggName()] =  + 
	}
	for ,  := range .GroupIdx {
		if ,  := [.Returns[].Name]; ! {
			[.Returns[].Name] = 
		}
	}
	 := make([]RowEval, len(.Post))
	for ,  := range .Post {
		[] = compileEval(, .Expr, )
	}
	return , 
}

// forEachGroup finalizes every group into a reused stride-wide scratch
// and hands the visible prefix to fn -- the streaming form of finalize
// for consumers that retain nothing per group (a keyless aggregate over
// no input still emits its zeroed row). The row is only valid for the
// duration of the call.
func ( *aggregator) ( *eval.Ctx,  *plan.ProjPlan,  func( []value.Value)) {
	if .nGroups == 0 && len(.GroupIdx) == 0 {
		.appendGroup(nil)
	}
	 := len(.Returns)
	,  := .postCompile(, )
	 :=  + .NHidden
	 := make([]value.Value, )
	for  := 0;  < .nGroups; ++ {
		clear()
		.emitGroup(, , , , , )
		([:])
	}
}

// finalize emits one row per group (a zeroed row for a keyless aggregate
// over no input), applies the nested-aggregate scalar wrappers over the
// hidden slots, then orders and paginates.
func ( *aggregator) ( *eval.Ctx,  *plan.ProjPlan,  map[string]int) [][]value.Value {
	if .nGroups == 0 && len(.GroupIdx) == 0 {
		.appendGroup(nil)
	}
	 := len(.Returns)
	,  := .postCompile(, )
	 :=  + .NHidden
	// The streamed path only pays off when groups can actually be
	// rejected; at nGroups <= bound it would build every row anyway, plus
	// heap bookkeeping the plain sort does not have.
	if  := orderBound();  >= 0 &&  < .nGroups && len(.OrderBy) > 0 && !.Distinct && !disableAggTopk {
		return .finalizeTopK(, , , , , , , )
	}
	// One arena backs every output row instead of a make per group: a
	// grouping over a million groups then pays one large allocation plus its
	// row-header slice, not a million small ones. Each row is a stride
	// window (nCols visible columns + the hidden accumulator slots the
	// post-wrappers read); only the visible prefix is published.
	 := make([]value.Value, .nGroups*)
	 := make([][]value.Value, 0, .nGroups)
	for  := 0;  < .nGroups; ++ {
		 := [* : *+ : *+]
		.emitGroup(, , , , , )
		 = append(, [:])
	}
	if len(.OrderBy) > 0 {
		 = sortRowsByOrder(, , , func(int) []value.Value { return nil }, 0, )
	}
	return paginate(, .Skip, .Limit)
}

// emitGroup assembles group idx's output row into row (stride wide: the
// visible columns plus the hidden accumulator slots the post-aggregation
// wrappers read).
func ( *aggregator) ( *eval.Ctx,  *plan.ProjPlan,  int,  []value.Value,  []RowEval,  map[string]int) {
	 := .keysOf()
	for ,  := range .GroupIdx {
		[] = []
	}
	 := .statesOf()
	var  []value.Value
	if .hasMinMax {
		 = .mmOf()
	}
	var  [][]value.Value
	if .hasCollect {
		 = .itemsOf()
	}
	for  := range .Aggs {
		switch .kinds[] {
		case plan.AggMin, plan.AggMax:
			// A null slot means no non-null arg was seen -> Null.
			[.Aggs[].OutIdx] = []
		case plan.AggCollect:
			[.Aggs[].OutIdx] = value.List([])
		case plan.AggPercentileCont, plan.AggPercentileDisc:
			[.Aggs[].OutIdx] = percentileOf(, .pctC[], [], .kinds[] == plan.AggPercentileCont)
		default:
			[.Aggs[].OutIdx] = [].finalize()
		}
	}
	for ,  := range .Post {
		[.Col] = [].Eval(, , )
	}
}

// finalizeTopK is finalize's bounded ORDER BY + LIMIT path: each group
// finalizes into a reused stride-wide scratch, its key vector is
// evaluated exactly as sortRowsByOrder would (output column, else the
// key expression over the finalized row under the column scope), and
// only rows the bounded accumulator would admit are copied out -- so at
// most bound group rows materialize instead of all of them, and the
// full-width sort decoration never exists. Selection equals
// sort-then-truncate by topKRows's total order (keys, then group
// arrival, matching the sort's index tiebreak).
func ( *aggregator) ( *eval.Ctx,  *plan.ProjPlan,  map[string]int,  []RowEval,  map[string]int, , ,  int) [][]value.Value {
	 := len(.OrderBy)
	 := make([]int, )
	for  := range .OrderBy {
		[] = plan.OrderColIndex(.OrderBy[].Expr, .Columns, .Returns)
	}
	 := make(map[string]int, len()+len(.Columns))
	maps.Copy(, )
	for ,  := range .Columns {
		[] = 
	}
	 := newTopKRows(, , .OrderBy)
	 := make([]value.Value, )
	 := make([]value.Value, )
	// Retention is bounded (plus eviction turnover), so size chunks to the
	// bound instead of paying a full-size chunk for a small LIMIT.
	 := rowArena{width: , chunkValues: min(arenaChunkValues, max(64, 2*)*)}
	for  := 0;  < .nGroups; ++ {
		clear()
		.emitGroup(, , , , , )
		for  := range .OrderBy {
			if  := [];  >= 0 {
				[] = []
			} else {
				[] = eval.Eval(, .OrderBy[].Expr, [:], )
			}
		}
		if !.wouldAccept() {
			.seq++ // rejected offers still order future arrivals
			continue
		}
		aggTopkBuilds++
		 := .alloc()
		copy(, [:])
		if !.offer(, ) {
			.rollback()
		}
	}
	return paginate(.sorted(), .Skip, .Limit)
}

// hiddenAggName is the rewritten hidden-slot variable a post-aggregation
// wrapper reads (must match the planner's __agg{k} rewrite).
func hiddenAggName( int) string {
	return "__agg" + strconv.Itoa()
}

// aggSink is the aggregated terminal sink: it streams matched rows into
// the group accumulator, so only per-group state is retained.
type aggSink struct {
	ctx   *eval.Ctx
	agg   *aggregator
	proj  *plan.ProjPlan
	slots map[string]int
}

func newAggSink( *eval.Ctx,  *plan.ProjPlan,  map[string]int) *aggSink {
	return &aggSink{ctx: , agg: newAggregator(, , ), proj: , slots: }
}

// push always reports true: aggregation must consume every row (the
// necessary asymmetry of the stop protocol).
func ( *aggSink) ( []value.Value) bool {
	.agg.update(.ctx, , .proj, .slots)
	return true
}

func ( *aggSink) () {}

func ( *aggSink) () [][]value.Value {
	return .agg.finalize(.ctx, .proj, .slots)
}

// percentileOf finalizes one percentile aggregate: sort the group's
// collected numeric values and pick per Neo4j semantics -- cont
// interpolates linearly between the two straddling values (always Float),
// disc takes the nearest-rank collected value unchanged. An empty group
// or a percentile outside [0,1] is Null.
func percentileOf( *eval.Ctx,  RowEval,  []value.Value,  bool) value.Value {
	if  == nil || len() == 0 {
		return value.Null()
	}
	,  := .Eval(, nil, nil).AsFloat()
	if ! ||  < 0 ||  > 1 {
		return value.Null()
	}
	slices.SortStableFunc(, func(,  value.Value) int {
		,  := .AsFloat()
		,  := .AsFloat()
		return cmp.Compare(, )
	})
	 := len()
	if ! {
		// Nearest rank: ceil(p*n) clamped to [1, n], 1-based.
		 := int(math.Ceil( * float64()))
		if  < 1 {
			 = 1
		}
		return [-1]
	}
	 :=  * float64(-1)
	 := int(math.Floor())
	 := int(math.Ceil())
	,  := [].AsFloat()
	,  := [].AsFloat()
	return value.Float( + (-)*(-float64()))
}