influxdb/query/point.go

package query

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"math"
	"sort"

	"github.com/gogo/protobuf/proto"
	internal "github.com/influxdata/influxdb/query/internal"
	"github.com/influxdata/influxql"
)

// ZeroTime is the Unix nanosecond timestamp for no time.
// This time is not used by the query engine or the storage engine as a valid time.
const ZeroTime = int64(math.MinInt64)

// Point represents a value in a series that occurred at a given time.
type Point interface {
	// Name and tags uniquely identify the series the value belongs to.
	name() string
	tags() Tags

	// The time that the value occurred at.
	time() int64

	// The value at the given time.
	value() interface{}

	// Auxillary values passed along with the value.
	aux() []interface{}
}

// Points represents a list of points.
type Points []Point

// Clone returns a deep copy of a.
func (a Points) Clone() []Point {
	other := make([]Point, len(a))
	for i, p := range a {
		if p == nil {
			other[i] = nil
			continue
		}

		switch p := p.(type) {
		case *FloatPoint:
			other[i] = p.Clone()
		case *IntegerPoint:
			other[i] = p.Clone()
		case *UnsignedPoint:
			other[i] = p.Clone()
		case *StringPoint:
			other[i] = p.Clone()
		case *BooleanPoint:
			other[i] = p.Clone()
		default:
			panic(fmt.Sprintf("unable to clone point: %T", p))
		}
	}
	return other
}

// Tags represent a map of keys and values.
// It memoizes its key so it can be used efficiently during query execution.
type Tags struct {
	id string
	m  map[string]string
}

// NewTags returns a new instance of Tags.
func NewTags(m map[string]string) Tags {
	if len(m) == 0 {
		return Tags{}
	}
	return Tags{
		id: string(encodeTags(m)),
		m:  m,
	}
}

// newTagsID returns a new instance of Tags by parsing the given tag ID.
func newTagsID(id string) Tags {
	m := decodeTags([]byte(id))
	if len(m) == 0 {
		return Tags{}
	}
	return Tags{id: id, m: m}
}

// Equal compares if the Tags are equal to each other.
func (t Tags) Equal(other Tags) bool {
	return t.ID() == other.ID()
}

// ID returns the string identifier for the tags.
func (t Tags) ID() string { return t.id }

// KeyValues returns the underlying map for the tags.
func (t Tags) KeyValues() map[string]string { return t.m }

// Keys returns a sorted list of all keys on the tag.
func (t *Tags) Keys() []string {
	if t == nil {
		return nil
	}

	var a []string
	for k := range t.m {
		a = append(a, k)
	}
	sort.Strings(a)
	return a
}

// Values returns a sorted list of all values on the tag.
func (t *Tags) Values() []string {
	if t == nil {
		return nil
	}

	a := make([]string, 0, len(t.m))
	for _, v := range t.m {
		a = append(a, v)
	}
	sort.Strings(a)
	return a
}

// Value returns the value for a given key.
func (t *Tags) Value(k string) string {
	if t == nil {
		return ""
	}
	return t.m[k]
}

// Subset returns a new tags object with a subset of the keys.
func (t *Tags) Subset(keys []string) Tags {
	if len(keys) == 0 {
		return Tags{}
	}

	// If keys match existing keys, simply return this tagset.
	if keysMatch(t.m, keys) {
		return *t
	}

	// Otherwise create new tag set.
	m := make(map[string]string, len(keys))
	for _, k := range keys {
		m[k] = t.m[k]
	}
	return NewTags(m)
}

// Equals returns true if t equals other.
func (t *Tags) Equals(other *Tags) bool {
	if t == nil && other == nil {
		return true
	} else if t == nil || other == nil {
		return false
	}
	return t.id == other.id
}

// keysMatch returns true if m has exactly the same keys as listed in keys.
func keysMatch(m map[string]string, keys []string) bool {
	if len(keys) != len(m) {
		return false
	}

	for _, k := range keys {
		if _, ok := m[k]; !ok {
			return false
		}
	}

	return true
}

// encodeTags converts a map of strings to an identifier.
func encodeTags(m map[string]string) []byte {
	// Empty maps marshal to empty bytes.
	if len(m) == 0 {
		return nil
	}

	// Extract keys and determine final size.
	sz := (len(m) * 2) - 1 // separators
	keys := make([]string, 0, len(m))
	for k, v := range m {
		keys = append(keys, k)
		sz += len(k) + len(v)
	}
	sort.Strings(keys)

	// Generate marshaled bytes.
	b := make([]byte, sz)
	buf := b
	for _, k := range keys {
		copy(buf, k)
		buf[len(k)] = '\x00'
		buf = buf[len(k)+1:]
	}
	for i, k := range keys {
		v := m[k]
		copy(buf, v)
		if i < len(keys)-1 {
			buf[len(v)] = '\x00'
			buf = buf[len(v)+1:]
		}
	}
	return b
}

// decodeTags parses an identifier into a map of tags.
func decodeTags(id []byte) map[string]string {
	a := bytes.Split(id, []byte{'\x00'})

	// There must be an even number of segments.
	if len(a) > 0 && len(a)%2 == 1 {
		a = a[:len(a)-1]
	}

	// Return nil if there are no segments.
	if len(a) == 0 {
		return nil
	}
	mid := len(a) / 2

	// Decode key/value tags.
	m := make(map[string]string)
	for i := 0; i < mid; i++ {
		m[string(a[i])] = string(a[i+mid])
	}
	return m
}

func encodeAux(aux []interface{}) []*internal.Aux {
	pb := make([]*internal.Aux, len(aux))
	for i := range aux {
		switch v := aux[i].(type) {
		case float64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Float)), FloatValue: proto.Float64(v)}
		case *float64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Float))}
		case int64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Integer)), IntegerValue: proto.Int64(v)}
		case *int64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Integer))}
		case uint64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Unsigned)), UnsignedValue: proto.Uint64(v)}
		case *uint64:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Unsigned))}
		case string:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.String)), StringValue: proto.String(v)}
		case *string:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.String))}
		case bool:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Boolean)), BooleanValue: proto.Bool(v)}
		case *bool:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Boolean))}
		default:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(influxql.Unknown))}
		}
	}
	return pb
}

func decodeAux(pb []*internal.Aux) []interface{} {
	if len(pb) == 0 {
		return nil
	}

	aux := make([]interface{}, len(pb))
	for i := range pb {
		switch influxql.DataType(pb[i].GetDataType()) {
		case influxql.Float:
			if pb[i].FloatValue != nil {
				aux[i] = *pb[i].FloatValue
			} else {
				aux[i] = (*float64)(nil)
			}
		case influxql.Integer:
			if pb[i].IntegerValue != nil {
				aux[i] = *pb[i].IntegerValue
			} else {
				aux[i] = (*int64)(nil)
			}
		case influxql.Unsigned:
			if pb[i].UnsignedValue != nil {
				aux[i] = *pb[i].UnsignedValue
			} else {
				aux[i] = (*uint64)(nil)
			}
		case influxql.String:
			if pb[i].StringValue != nil {
				aux[i] = *pb[i].StringValue
			} else {
				aux[i] = (*string)(nil)
			}
		case influxql.Boolean:
			if pb[i].BooleanValue != nil {
				aux[i] = *pb[i].BooleanValue
			} else {
				aux[i] = (*bool)(nil)
			}
		default:
			aux[i] = nil
		}
	}
	return aux
}

func cloneAux(src []interface{}) []interface{} {
	if src == nil {
		return src
	}
	dest := make([]interface{}, len(src))
	copy(dest, src)
	return dest
}

// PointDecoder decodes generic points from a reader.
type PointDecoder struct {
	r     io.Reader
	stats IteratorStats
}

// NewPointDecoder returns a new instance of PointDecoder that reads from r.
func NewPointDecoder(r io.Reader) *PointDecoder {
	return &PointDecoder{r: r}
}

// Stats returns iterator stats embedded within the stream.
func (dec *PointDecoder) Stats() IteratorStats { return dec.stats }

// DecodePoint reads from the underlying reader and unmarshals into p.
func (dec *PointDecoder) DecodePoint(p *Point) error {
	for {
		// Read length.
		var sz uint32
		if err := binary.Read(dec.r, binary.BigEndian, &sz); err != nil {
			return err
		}

		// Read point data.
		buf := make([]byte, sz)
		if _, err := io.ReadFull(dec.r, buf); err != nil {
			return err
		}

		// Unmarshal into point.
		var pb internal.Point
		if err := proto.Unmarshal(buf, &pb); err != nil {
			return err
		}

		// If the point contains stats then read stats and retry.
		if pb.Stats != nil {
			dec.stats = decodeIteratorStats(pb.Stats)
			continue
		}

		if pb.IntegerValue != nil {
			*p = decodeIntegerPoint(&pb)
		} else if pb.UnsignedValue != nil {
			*p = decodeUnsignedPoint(&pb)
		} else if pb.StringValue != nil {
			*p = decodeStringPoint(&pb)
		} else if pb.BooleanValue != nil {
			*p = decodeBooleanPoint(&pb)
		} else {
			*p = decodeFloatPoint(&pb)
		}

		return nil
	}
}