// processAggregateNodeOutputs adds the aggregated outputs of a withItems/withParam template as a
// parameter in the form of a JSON list
420489
func (woc *wfOperationCtx) processAggregateNodeOutputs(templateName string, scope *wfScope, prefix string, childNodes []wfv1.NodeStatus) {
	if len(childNodes) == 0 {
		return
	}
	// need to sort the child node list so that the order of outputs are preserved
	sort.Sort(loopNodes(childNodes))
	paramList := make([]map[string]string, 0)
	resultsList := make([]wfv1.Item, 0)
	for _, node := range childNodes {
		if node.Outputs == nil {
			continue
		}
		if len(node.Outputs.Parameters) > 0 {
			param := make(map[string]string)
			for _, p := range node.Outputs.Parameters {
				param[p.Name] = *p.Value
			}
			paramList = append(paramList, param)
		}
		if node.Outputs.Result != nil {
			// Support the case where item may be a map
			var itemMap map[string]interface{}
			err := json.Unmarshal([]byte(*node.Outputs.Result), &itemMap)
			if err == nil {
				resultsList = append(resultsList, wfv1.Item{Value: itemMap})
			} else {
				resultsList = append(resultsList, wfv1.Item{Value: *node.Outputs.Result})
			}
		}
	}
	tmplType := woc.wf.GetTemplate(templateName).GetType()
	if tmplType == wfv1.TemplateTypeScript {
		resultsJSON, _ := json.Marshal(resultsList)
		key := fmt.Sprintf("%s.outputs.result", prefix)
		scope.addParamToScope(key, string(resultsJSON))
	}
	outputsJSON, _ := json.Marshal(paramList)
	key := fmt.Sprintf("%s.outputs.parameters", prefix)
	scope.addParamToScope(key, string(outputsJSON))
}
510365
func processArg(arg *Arg) (*Arg, error) {
	osArg := libInputOutput.readOSArg()
	// default parameters - can be parsed as JSON
	if osArg == "" {
		osArg = "{}"
	}
	err := json.Unmarshal([]byte(osArg), arg)
	if err != nil {
		return nil, err
	}

	return arg, nil
}
694016
func (b *builder) processFilterNode(root *filterNode) (query, error) {
	b.flag |= filterFlag

	qyInput, err := b.processNode(root.Input)
	if err != nil {
		return nil, err
	}
	qyCond, err := b.processNode(root.Condition)
	if err != nil {
		return nil, err
	}
	qyOutput := &filterQuery{Input: qyInput, Predicate: qyCond}
	return qyOutput, nil
}
144458
func parseOutputCSV(s string) (client.ExportEntry, error) {
	ex := client.ExportEntry{
		Type:  "",
		Attrs: map[string]string{},
	}
	csvReader := csv.NewReader(strings.NewReader(s))
	fields, err := csvReader.Read()
	if err != nil {
		return ex, err
	}
	for _, field := range fields {
		parts := strings.SplitN(field, "=", 2)
		if len(parts) != 2 {
			return ex, errors.Errorf("invalid value %s", field)
		}
		key := strings.ToLower(parts[0])
		value := parts[1]
		switch key {
		case "type":
			ex.Type = value
		default:
			ex.Attrs[key] = value
		}
	}
	if ex.Type == "" {
		return ex, errors.New("--output requires type=<type>")
	}
	if v, ok := ex.Attrs["output"]; ok {
		return ex, errors.Errorf("output=%s not supported for --output, you meant dest=%s?", v, v)
	}
	ex.Output, ex.OutputDir, err = resolveExporterDest(ex.Type, ex.Attrs["dest"])
	if err != nil {
		return ex, errors.Wrap(err, "invalid output option: output")
	}
	if ex.Output != nil || ex.OutputDir != "" {
		delete(ex.Attrs, "dest")
	}
	return ex, nil
}
166251
func (a *FiltersAggregation) SubAggregation(name string, subAggregation Aggregation) *FiltersAggregation {
	a.subAggregations[name] = subAggregation
	return a
}
1262315
func (dg *discoveryGateway) GetAggregateStats(target *querypb.Target) (*querypb.AggregateStats, queryservice.QueryService, error) {
	stats, err := dg.tsc.GetAggregateStats(target)
	return stats, dg, err
}
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
}
403635
func (cn *commentNode) ProcessWith(p Processor) error {
	return p.Comment(cn.comment)
}
1355255
func getOutput(dagnode ipld.Node) (*coreiface.AddEvent, error) {
	c := dagnode.Cid()
	s, err := dagnode.Size()
	if err != nil {
		return nil, err
	}

	output := &coreiface.AddEvent{
		Path: path.IpfsPath(c),
		Size: strconv.FormatUint(s, 10),
	}

	return output, nil
}
906630
func ioOutputFormatFunction(ld *LineData, b *bytes.Buffer) error {
	_, err := b.WriteString(ld.CallingFunction)
	return err
}
539248
func (a *HistoricalApi) nodeAggregations(request *restful.Request, response *restful.Response) {
	key := core.HistoricalKey{
		ObjectType: core.MetricSetTypeNode,
		NodeName:   request.PathParameter("node-name"),
	}
	a.processAggregationRequest(key, request, response)
}
724470
func (def *Synthdef) processUgenInput(input Input, stack *stack, depth int) {
	switch v := input.(type) {
	case *Ugen:
		def.topsortr(v, stack, depth+1)
		break
	case MultiInput:
		// multi input
		mins := v.InputArray()
		for j := len(mins) - 1; j >= 0; j-- {
			switch w := mins[j].(type) {
			case *Ugen:
				def.topsortr(w, stack, depth+1)
				break
			}
		}
		break
	}
}
846886
func (c *LogCmd) CombinedOutput() ([]byte, error) {
	st := time.Now()
	data, err := c.Cmd.CombinedOutput()
	c.log(st, err, nil)
	return data, err
}
1259991
func ParsePSOutput(output []byte) (*TopResults, error) {
	topResults := &TopResults{}

	lines := strings.Split(string(output), "\n")
	topResults.Headers = fieldsASCII(lines[0])

	pidIndex := -1
	for i, name := range topResults.Headers {
		if name == "PID" {
			pidIndex = i
		}
	}

	for _, line := range lines[1:] {
		if len(line) == 0 {
			continue
		}

		fields := fieldsASCII(line)

		if fields[pidIndex] == "-" {
			continue
		}

		process := fields[:len(topResults.Headers)-1]
		process = append(process, strings.Join(fields[len(topResults.Headers)-1:], " "))
		topResults.Processes = append(topResults.Processes, process)

	}
	return topResults, nil
}
192289
func processJaegerTags(s *jThrift.Span) (*string, *trace.Endpoint, map[string]string) {
	var kind *string
	var remote *trace.Endpoint
	tags := make(map[string]string, len(s.Tags))

	ensureRemote := func() {
		if remote == nil {
			remote = &trace.Endpoint{}
		}
	}

	for i := range s.Tags {
		switch s.Tags[i].Key {
		case string(ext.PeerHostIPv4):
			ip := convertPeerIPv4(s.Tags[i])
			if ip == "" {
				continue
			}
			ensureRemote()
			remote.Ipv4 = pointer.String(ip)
		// ipv6 host is always string
		case string(ext.PeerHostIPv6):
			if s.Tags[i].VStr != nil {
				ensureRemote()
				remote.Ipv6 = s.Tags[i].VStr
			}
		case string(ext.PeerPort):
			port := convertPeerPort(s.Tags[i])
			if port == 0 {
				continue
			}
			ensureRemote()
			remote.Port = &port
		case string(ext.PeerService):
			ensureRemote()
			remote.ServiceName = s.Tags[i].VStr
		case string(ext.SpanKind):
			kind = convertKind(s.Tags[i])
		default:
			val := tagValueToString(s.Tags[i])
			if val != "" {
				tags[s.Tags[i].Key] = val
			}
		}
	}
	return kind, remote, tags
}
315178
func HookOutput(outputs []*bytes.Buffer) []byte {
	r := []byte{}
	for _, m := range outputs {
		r = append(r, m.Bytes()...)
	}
	return r
}
518509
func FlagOutput(in worker.Worker, out interface{}) error {
	inFlag, ok := in.(Flag)
	if !ok {
		return errors.Errorf("expected in to implement Flag; got a %T", in)
	}
	outFlag, ok := out.(*Flag)
	if !ok {
		return errors.Errorf("expected out to be a *Flag; got a %T", out)
	}
	*outFlag = inFlag
	return nil
}