// aggPushDown tries to push down aggregate functions to join paths.
353397
func (a *aggPushDownSolver) aggPushDown(p LogicalPlan) {
	if agg, ok := p.(*Aggregation); ok {
		child := agg.GetChildByIndex(0)
		if join, ok1 := child.(*Join); ok1 && a.checkValidJoin(join) {
			if valid, leftAggFuncs, rightAggFuncs, leftGbyCols, rightGbyCols := a.splitAggFuncsAndGbyCols(agg, join); valid {
				var lChild, rChild LogicalPlan
				// If there exist count or sum functions in left join path, we can't push any
				// aggregate function into right join path.
				rightInvalid := a.checkAnyCountAndSum(leftAggFuncs)
				leftInvalid := a.checkAnyCountAndSum(rightAggFuncs)
				if rightInvalid {
					rChild = join.GetChildByIndex(1).(LogicalPlan)
				} else {
					rChild = a.tryToPushDownAgg(rightAggFuncs, rightGbyCols, join, 1)
				}
				if leftInvalid {
					lChild = join.GetChildByIndex(0).(LogicalPlan)
				} else {
					lChild = a.tryToPushDownAgg(leftAggFuncs, leftGbyCols, join, 0)
				}
				join.SetChildren(lChild, rChild)
				lChild.SetParents(join)
				rChild.SetParents(join)
				join.SetSchema(expression.MergeSchema(lChild.GetSchema(), rChild.GetSchema()))
			}
		} else if proj, ok1 := child.(*Projection); ok1 {
			// TODO: This optimization is not always reasonable. We have not supported pushing projection to kv layer yet,
			// so we must do this optimization.
			for i, gbyItem := range agg.GroupByItems {
				agg.GroupByItems[i] = expression.ColumnSubstitute(gbyItem, proj.schema, proj.Exprs)
			}
			agg.collectGroupByColumns()
			for _, aggFunc := range agg.AggFuncs {
				newArgs := make([]expression.Expression, 0, len(aggFunc.GetArgs()))
				for _, arg := range aggFunc.GetArgs() {
					newArgs = append(newArgs, expression.ColumnSubstitute(arg, proj.schema, proj.Exprs))
				}
				aggFunc.SetArgs(newArgs)
			}
			projChild := proj.children[0]
			agg.SetChildren(projChild)
			projChild.SetParents(agg)
		} else if union, ok1 := child.(*Union); ok1 {
			pushedAgg := a.makeNewAgg(agg.AggFuncs, agg.groupByCols)
			newChildren := make([]Plan, 0, len(union.children))
			for _, child := range union.children {
				newChild := a.pushAggCrossUnion(pushedAgg, union.schema, child.(LogicalPlan))
				newChildren = append(newChildren, newChild)
				newChild.SetParents(union)
			}
			union.SetChildren(newChildren...)
			union.SetSchema(pushedAgg.schema)
		}
	}
	for _, child := range p.GetChildren() {
		a.aggPushDown(child.(LogicalPlan))
	}
}
809001
func AggMax(cols ...string) Aggregator {
	max := func(xs []float64) float64 {
		_, x := stats.Bounds(xs)
		return x
	}
	return aggFn(max, "max ", cols...)
}
181656
func NewEventForAggregate(eventType EventType, data EventData, timestamp time.Time,
	aggregateType AggregateType, aggregateID uuid.UUID, version int) Event {
	return event{
		eventType:     eventType,
		data:          data,
		timestamp:     timestamp,
		aggregateType: aggregateType,
		aggregateID:   aggregateID,
		version:       version,
	}
}
504508
func NewAggregate(errs ...error) error {
	// filter out possible nil values
	var nonNils []error
	for _, err := range errs {
		if err != nil {
			nonNils = append(nonNils, err)
		}
	}
	if len(nonNils) == 0 {
		return nil
	}
	return wrapWithDepth(aggregate(nonNils), 2)
}
565581
func (e *StreamAggExec) appendResult2Chunk(chk *chunk.Chunk) error {
	for i, aggFunc := range e.aggFuncs {
		err := aggFunc.AppendFinalResult2Chunk(e.ctx, e.partialResults[i], chk)
		if err != nil {
			return err
		}
		aggFunc.ResetPartialResult(e.partialResults[i])
	}
	if len(e.aggFuncs) == 0 {
		chk.SetNumVirtualRows(chk.NumRows() + 1)
	}
	return nil
}
26821
func (s *Store) FindAggregate(dataType interface{}, query *Query, groupBy ...string) ([]*AggregateResult, error) {
	var result []*AggregateResult
	var err error
	err = s.Bolt().View(func(tx *bolt.Tx) error {
		result, err = s.TxFindAggregate(tx, dataType, query, groupBy...)
		return err
	})

	if err != nil {
		return nil, err
	}

	return result, nil
}
548474
func (c *syncClient) Lpush(arg0 string, arg1 []byte) (err Error) {
	arg0bytes := []byte(arg0)
	arg1bytes := arg1

	_, err = c.conn.ServiceRequest(&LPUSH, [][]byte{arg0bytes, arg1bytes})
	return
}
166251
func (a *FiltersAggregation) SubAggregation(name string, subAggregation Aggregation) *FiltersAggregation {
	a.subAggregations[name] = subAggregation
	return a
}
566192
func (a *CumulativeSumAggregation) BucketsPath(bucketsPaths ...string) *CumulativeSumAggregation {
	a.bucketsPaths = append(a.bucketsPaths, bucketsPaths...)
	return a
}
1057778
func (d *AggregateDsl) Aggregates(aggs ...*AggregateDsl) *AggregateDsl {
	if len(aggs) < 1 {
		return d
	}
	if len(d.AggregatesVal) == 0 {
		d.AggregatesVal = make(map[string]*AggregateDsl)
	}

	for _, agg := range aggs {
		d.AggregatesVal[agg.Name] = agg
	}
	return d
}
969043
func (s *SearchService) Aggregation(name string, aggregation Aggregation) *SearchService {
	s.searchSource = s.searchSource.Aggregation(name, aggregation)
	return s
}
386802
func (r *SearchRequest) Aggregation(name string, aggregation Aggregation) *SearchRequest {
	r.searchSource = r.searchSource.Aggregation(name, aggregation)
	return r
}
1188142
func (agg *Aggregator) flush() {
	if agg.minMetric != nil {
		agg.minMetric.Add(agg.currentBoundary, agg.agg.Min)
	}
	if agg.maxMetric != nil {
		agg.maxMetric.Add(agg.currentBoundary, agg.agg.Max)
	}
	if agg.sumMetric != nil {
		agg.sumMetric.Add(agg.currentBoundary, agg.agg.Sum)
	}
	if agg.cntMetric != nil {
		agg.cntMetric.Add(agg.currentBoundary, agg.agg.Cnt)
	}
	if agg.lstMetric != nil {
		agg.lstMetric.Add(agg.currentBoundary, agg.agg.Lst)
	}
	//msg := fmt.Sprintf("flushed cnt %v sum %f min %f max %f, reset the block", agg.agg.cnt, agg.agg.sum, agg.agg.min, agg.agg.max)
	agg.agg.Reset()
}
52920
func (tc *TabletStatsCache) GetAggregateStats(target *querypb.Target) (*querypb.AggregateStats, error) {
	e := tc.getEntry(target.Keyspace, target.Shard, target.TabletType)
	if e == nil {
		return nil, topo.NewError(topo.NoNode, topotools.TargetIdent(target))
	}

	e.mu.RLock()
	defer e.mu.RUnlock()
	if target.TabletType == topodatapb.TabletType_MASTER {
		if len(e.aggregates) == 0 {
			return nil, topo.NewError(topo.NoNode, topotools.TargetIdent(target))
		}
		for _, agg := range e.aggregates {
			return agg, nil
		}
	}
	targetAlias := tc.getAliasByCell(target.Cell)
	agg, ok := e.aggregates[targetAlias]
	if !ok {
		return nil, topo.NewError(topo.NoNode, topotools.TargetIdent(target))
	}
	return agg, nil
}
1152069
func (a Aggregations) Global(name string) (*AggregationSingleBucket, bool) {
	if raw, found := a[name]; found {
		agg := new(AggregationSingleBucket)
		if raw == nil {
			return agg, true
		}
		if err := json.Unmarshal(raw, agg); err == nil {
			return agg, true
		}
	}
	return nil, false
}
542976
func (e *FuncExpr) evalAggregationNode(r Record) error {
	// It is assumed that this function is called only when
	// `e` is an aggregation function.

	var val *Value
	var err error
	funcName := e.getFunctionName()
	if aggFnCount == funcName {
		if e.Count.StarArg {
			// Handle COUNT(*)
			e.aggregate.runningCount++
			return nil
		}

		val, err = e.Count.ExprArg.evalNode(r)
		if err != nil {
			return err
		}
	} else {
		// Evaluate the (only) argument
		val, err = e.SFunc.ArgsList[0].evalNode(r)
		if err != nil {
			return err
		}
	}

	if val.IsNull() {
		// E.g. the column or field does not exist in the
		// record - in all such cases the aggregation is not
		// updated.
		return nil
	}

	argVal := val
	if funcName != aggFnCount {
		// All aggregation functions, except COUNT require a
		// numeric argument.

		// Here, we diverge from Amazon S3 behavior by
		// inferring untyped values are numbers.
		if i, ok := argVal.bytesToInt(); ok {
			argVal.setInt(i)
		} else if f, ok := argVal.bytesToFloat(); ok {
			argVal.setFloat(f)
		} else {
			return errNonNumericArg(funcName)
		}
	}

	// Mark that we have seen one non-null value.
	isFirstRow := false
	if !e.aggregate.seen {
		e.aggregate.seen = true
		isFirstRow = true
	}

	switch funcName {
	case aggFnCount:
		// For all non-null values, the count is incremented.
		e.aggregate.runningCount++

	case aggFnAvg:
		e.aggregate.runningCount++
		err = e.aggregate.runningSum.arithOp(opPlus, argVal)

	case aggFnMin:
		err = e.aggregate.runningMin.minmax(argVal, false, isFirstRow)

	case aggFnMax:
		err = e.aggregate.runningMax.minmax(argVal, true, isFirstRow)

	case aggFnSum:
		err = e.aggregate.runningSum.arithOp(opPlus, argVal)

	default:
		err = errInvalidAggregation
	}

	return err
}
643427
func (list ErrorList) ToAggregate() utilerrors.Aggregate {
	errs := make([]error, len(list))
	for i := range list {
		errs[i] = list[i]
	}
	return utilerrors.NewAggregate(errs)
}