influxql.gen.go.tmpl

package kapacitor


import (
	"fmt"
	"time"

	"github.com/influxdata/influxdb/influxql"
	"github.com/influxdata/kapacitor/models"
	"github.com/influxdata/kapacitor/pipeline"
)

{{/* Define typed Aggregate/Emit types */}}
{{range .}}

type {{.name}}PointAggregator struct {
	field         string
	topBottomInfo *pipeline.TopBottomCallInfo
	isSimpleSelector bool
	aggregator influxql.{{.Name}}PointAggregator
}

func {{.name}}PopulateAuxFieldsAndTags(ap *influxql.{{.Name}}Point, fieldsAndTags []string, fields models.Fields, tags models.Tags) {
	ap.Aux = make([]interface{}, len(fieldsAndTags))
	for i, name := range fieldsAndTags {
		if f, ok := fields[name]; ok {
			ap.Aux[i] = f
		} else {
			ap.Aux[i] = tags[name]
		}
	}
}

func (a *{{.name}}PointAggregator) AggregateBatch(b *models.Batch) error {
	for _, p := range b.Points {
		value, ok := p.Fields[a.field]
		if !ok {
			return fmt.Errorf("field %s missing from point cannot aggregate", a.field)
		}
		typed, ok := value.({{.Type}})
		if !ok {
			return fmt.Errorf("field %s has wrong type: got %T exp {{.Type}}", a.field, value)
		}
		ap := &influxql.{{.Name}}Point{
			Name:  b.Name,
			Tags:  influxql.NewTags(p.Tags),
			Time:  p.Time.UnixNano(),
			Value: typed,
		}
		if a.topBottomInfo != nil {
			// We need to populate the Aux fields
			{{.name}}PopulateAuxFieldsAndTags(ap, a.topBottomInfo.FieldsAndTags, p.Fields, p.Tags)
		}

		if a.isSimpleSelector {
			ap.Aux = []interface{}{ p.Tags, p.Fields }
		}
		
		a.aggregator.Aggregate{{.Name}}(ap)
	}
	return nil
}

func (a *{{.name}}PointAggregator) AggregatePoint(p *models.Point) error {
	value, ok := p.Fields[a.field]
	if !ok {
		return fmt.Errorf("field %s missing from point cannot aggregate", a.field)
	}
	typed, ok := value.({{.Type}})
	if !ok {
		return fmt.Errorf("field %s has wrong type: got %T exp {{.Type}}", a.field, value)
	}
	ap := &influxql.{{.Name}}Point{
		Name:  p.Name,
		Tags:  influxql.NewTags(p.Tags),
		Time:  p.Time.UnixNano(),
		Value: typed,
	}
	if a.topBottomInfo != nil {
		// We need to populate the Aux fields
		{{.name}}PopulateAuxFieldsAndTags(ap, a.topBottomInfo.FieldsAndTags, p.Fields, p.Tags)
	}

	if a.isSimpleSelector {
		ap.Aux = []interface{}{ p.Tags, p.Fields }
	}
	
	a.aggregator.Aggregate{{.Name}}(ap)
	return nil
}



type {{.name}}PointBulkAggregator struct {
	field      string
	topBottomInfo *pipeline.TopBottomCallInfo
	isSimpleSelector bool
	aggregator    pipeline.{{.Name}}BulkPointAggregator
}

func (a *{{.name}}PointBulkAggregator) AggregateBatch(b *models.Batch) error {
	slice := make([]influxql.{{.Name}}Point, len(b.Points))
	for i, p := range b.Points {
		value, ok := p.Fields[a.field]
		if !ok {
			return fmt.Errorf("field %s missing from point cannot aggregate", a.field)
		}
		typed, ok := value.({{.Type}})
		if !ok {
			return fmt.Errorf("field %s has wrong type: got %T exp {{.Type}}", a.field, value)
		}
		slice[i] = influxql.{{.Name}}Point{
			Name:  b.Name,
			Tags:  influxql.NewTags(p.Tags),
			Time:  p.Time.UnixNano(),
			Value: typed,
		}
		if a.topBottomInfo != nil {
			// We need to populate the Aux fields
			{{.name}}PopulateAuxFieldsAndTags(&slice[i], a.topBottomInfo.FieldsAndTags, p.Fields, p.Tags)
		}

		if a.isSimpleSelector {
			slice[i].Aux = []interface{}{ p.Tags, p.Fields }
		}
	}
	a.aggregator.Aggregate{{.Name}}Bulk(slice)
	return nil
}

func (a *{{.name}}PointBulkAggregator) AggregatePoint(p *models.Point) error {
	value, ok := p.Fields[a.field]
	if !ok {
		return fmt.Errorf("field %s missing from point cannot aggregate", a.field)
	}
	typed, ok := value.({{.Type}})
	if !ok {
		return fmt.Errorf("field %s has wrong type: got %T exp {{.Type}}", a.field, value)
	}
	ap := &influxql.{{.Name}}Point{
		Name:  p.Name,
		Tags:  influxql.NewTags(p.Tags),
		Time:  p.Time.UnixNano(),
		Value: typed,
	}
	if a.topBottomInfo != nil {
		// We need to populate the Aux fields
		{{.name}}PopulateAuxFieldsAndTags(ap, a.topBottomInfo.FieldsAndTags, p.Fields, p.Tags)
	}

	if a.isSimpleSelector {
		ap.Aux = []interface{}{ p.Tags, p.Fields }
	}

	a.aggregator.Aggregate{{.Name}}(ap)
	return nil
}

type {{.name}}PointEmitter struct {
	baseReduceContext
	emitter influxql.{{.Name}}PointEmitter
	isSimpleSelector bool
    byName bool
}

func (e *{{.name}}PointEmitter) EmitPoint() (models.Point, error) {
	slice := e.emitter.Emit()
	if len(slice) != 1 {
		return models.Point{}, ErrEmptyEmit
	}
	ap := slice[0]
	var t time.Time
	if e.pointTimes {
		if ap.Time == influxql.ZeroTime {
			t = e.time
		} else {
			t = time.Unix(0, ap.Time).UTC()
		}
	} else {
		t = e.time
	}

	var fields models.Fields
	var tags models.Tags
	if e.isSimpleSelector {
		tags = ap.Aux[0].(models.Tags)
		fields = ap.Aux[1].(models.Fields)
		if e.as != e.field {
			fields = fields.Copy()
			fields[e.as] = fields[e.field]
			delete(fields, e.field)
		}
	} else {
		tags = e.tags
		fields = map[string]interface{}{e.as: ap.Value}
	}

	return models.Point{
		Name:       e.name,
		Time:       t,
		Group:      e.group,
		Dimensions: e.dimensions,
		Tags:       tags,
		Fields:     fields,
	}, nil
}

func (e *{{.name}}PointEmitter) EmitBatch() models.Batch {
	slice := e.emitter.Emit()
	b := models.Batch{
		Name:   e.name,
		TMax:   e.time,
		Group:  e.group,
		ByName: e.dimensions.ByName,
		Tags:   e.tags,
		Points: make([]models.BatchPoint, len(slice)),
	}
	var t time.Time
	for i, ap := range slice {
		if e.pointTimes {
			if ap.Time == influxql.ZeroTime {
				t = e.time
			} else {
				t = time.Unix(0, ap.Time).UTC()
			}
		} else {
			t = e.time
		}
		var tags models.Tags
		if l := len(ap.Tags.KeyValues()); l > 0 {
			// Merge batch and point specific tags
			tags = make(models.Tags, len(e.tags)+l)
			for k, v := range e.tags {
				tags[k] = v
			}
			for k, v := range ap.Tags.KeyValues() {
				tags[k] = v
			}
		} else {
			tags = e.tags
		}
		b.Points[i] = models.BatchPoint{
			Time:   t,
			Tags:   tags,
			Fields: map[string]interface{}{e.as: ap.Value},
		}
		if t.After(b.TMax) {
			b.TMax = t
		}
	}
	return b
}

{{end}}

{{/* Define composite types for reduceContext */}}
{{with $types := .}}
{{range $a := $types}}
{{range $e := $types}}

// {{$a.name}}{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext uses composition to implement the reduceContext interface
type {{$a.name}}{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext struct {
    {{$a.name}}PointAggregator
    {{$e.name}}PointEmitter
}

// {{$a.name}}Bulk{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext uses composition to implement the reduceContext interface
type {{$a.name}}Bulk{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext struct {
    {{$a.name}}PointBulkAggregator
    {{$e.name}}PointEmitter
}
{{end}}{{end}}


{{/* Define switch cases for reduceContext contruction */}}

func determineReduceContextCreateFn(method string, value interface{}, rc pipeline.ReduceCreater)  (fn createReduceContextFunc, err error) {
	switch value.(type) {
{{range $a := $types}}
	case {{.Type}}:
		switch {
{{range $e := $types}}
		case rc.Create{{$a.Name}}{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}Reducer != nil:
			 fn = func(c baseReduceContext) reduceContext {
				a, e := rc.Create{{$a.Name}}{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}Reducer()
				return &{{$a.name}}{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext{
					{{$a.name}}PointAggregator: {{$a.name}}PointAggregator{
						field:      c.field,
						topBottomInfo: rc.TopBottomCallInfo,
						isSimpleSelector: rc.IsSimpleSelector,
						aggregator: a,
					},
					{{$e.name}}PointEmitter: {{$e.name}}PointEmitter{
						baseReduceContext: c,
						emitter:           e,
						isSimpleSelector: rc.IsSimpleSelector,
					},
				}
			}
		case rc.Create{{$a.Name}}Bulk{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}Reducer != nil:
			 fn = func(c baseReduceContext) reduceContext {
				a, e := rc.Create{{$a.Name}}Bulk{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}Reducer()
				return &{{$a.name}}Bulk{{if ne $a.Name $e.Name}}{{$e.Name}}{{end}}ReduceContext{
					{{$a.name}}PointBulkAggregator: {{$a.name}}PointBulkAggregator{
						field:      c.field,
						topBottomInfo: rc.TopBottomCallInfo,
						isSimpleSelector: rc.IsSimpleSelector,
						aggregator: a,
					},
					{{$e.name}}PointEmitter: {{$e.name}}PointEmitter{
						baseReduceContext: c,
						emitter:           e,
						isSimpleSelector: rc.IsSimpleSelector,
					},
				}
			}
{{end}}
		default:
			err = fmt.Errorf("cannot apply %s to {{$a.Type}} field", method)
		}
{{end}}
	default:
		err = fmt.Errorf("invalid field type: %T", value)
	}
	return
}
{{end}}