// convertDatum returns a Go primitive value equivalent of val, of the
// type expected by col. If val's type is incompatible with col, or if
// col's type is not yet implemented, an error is returned.
func convertDatum(col ColumnDescriptor, val parser.Datum) (interface{}, error) {
	if val == parser.DNull {
		return nil, nil
	}

	switch col.Type.Kind {
	case ColumnType_BOOL:
		if v, ok := val.(parser.DBool); ok {
			return bool(v), nil
		}
	case ColumnType_INT:
		if v, ok := val.(parser.DInt); ok {
			return int64(v), nil
		}
	case ColumnType_FLOAT:
		if v, ok := val.(parser.DFloat); ok {
			return float64(v), nil
		}
	// case ColumnType_DECIMAL:
	// case ColumnType_DATE:
	// case ColumnType_TIMESTAMP:
	case ColumnType_STRING, ColumnType_BYTES:
		if v, ok := val.(parser.DString); ok {
			return string(v), nil
		}
	default:
		return nil, fmt.Errorf("unsupported type: %s", val.Type())
	}
	return nil, fmt.Errorf("value type %s doesn't match type %s of column %q",
		val.Type(), col.Type.Kind, col.Name)
}

// Encodes datum at the end of key, using direction `dir` for the encoding.
// The key is a span end key, which is exclusive, but `val` needs to
// be inclusive. So if datum is the last end constraint, we transform it accordingly.
func encodeInclusiveEndValue(
	key roachpb.Key, datum parser.Datum, dir encoding.Direction,
	isLastEndConstraint bool) roachpb.Key {
	// Since the end of a span is exclusive, if the last constraint is an
	// inclusive one, we might need to make the key exclusive by applying a
	// PrefixEnd().  We normally avoid doing this by transforming "a = x" to
	// "a = x±1" for the last end constraint, depending on the encoding direction
	// (since this keeps the key nice and pretty-printable).
	// However, we might not be able to do the ±1.
	needExclusiveKey := false
	if isLastEndConstraint {
		if dir == encoding.Ascending {
			if datum.IsMax() {
				needExclusiveKey = true
			} else {
				datum = datum.Next()
			}
		} else {
			if datum.IsMin() || !datum.HasPrev() {
				needExclusiveKey = true
			} else {
				datum = datum.Prev()
			}
		}
	}
	key, pErr := encodeTableKey(key, datum, dir)
	if pErr != nil {
		panic(pErr)
	}
	if needExclusiveKey {
		key = key.PrefixEnd()
	}
	return key
}

func decodeTableKey(valType parser.Datum, key []byte) (parser.Datum, []byte, error) {
	var isNull bool
	if key, isNull = encoding.DecodeIfNull(key); isNull {
		return parser.DNull, key, nil
	}
	switch valType.(type) {
	case parser.DBool:
		rkey, i, err := encoding.DecodeVarint(key)
		return parser.DBool(i != 0), rkey, err
	case parser.DInt:
		rkey, i, err := encoding.DecodeVarint(key)
		return parser.DInt(i), rkey, err
	case parser.DFloat:
		rkey, f, err := encoding.DecodeFloat(key, nil)
		return parser.DFloat(f), rkey, err
	case parser.DString:
		rkey, r, err := encoding.DecodeString(key, nil)
		return parser.DString(r), rkey, err
	case parser.DBytes:
		rkey, r, err := encoding.DecodeString(key, nil)
		return parser.DBytes(r), rkey, err
	case parser.DDate:
		rkey, t, err := encoding.DecodeTime(key)
		return parser.DDate{Time: t}, rkey, err
	case parser.DTimestamp:
		rkey, t, err := encoding.DecodeTime(key)
		return parser.DTimestamp{Time: t}, rkey, err
	case parser.DInterval:
		rkey, d, err := encoding.DecodeVarint(key)
		return parser.DInterval{Duration: time.Duration(d)}, rkey, err
	default:
		return nil, nil, util.Errorf("TODO(pmattis): decoded index key: %s", valType.Type())
	}
}

func decodeTableKey(valType parser.Datum, key []byte) (parser.Datum, []byte, error) {
	var isNull bool
	if key, isNull = encoding.DecodeIfNull(key); isNull {
		return parser.DNull, key, nil
	}
	switch valType.(type) {
	case parser.DBool:
		var i int64
		key, i = encoding.DecodeVarint(key)
		return parser.DBool(i != 0), key, nil
	case parser.DInt:
		var i int64
		key, i = encoding.DecodeVarint(key)
		return parser.DInt(i), key, nil
	case parser.DFloat:
		var f float64
		key, f = encoding.DecodeFloat(key, nil)
		return parser.DFloat(f), key, nil
	case parser.DString:
		var r string
		key, r = encoding.DecodeString(key, nil)
		return parser.DString(r), key, nil
	default:
		return nil, nil, util.Errorf("TODO(pmattis): decoded index key: %s", valType.Type())
	}
}

func (a *varianceAggregate) add(datum parser.Datum) error {
	if datum == parser.DNull {
		return nil
	}

	var d parser.DFloat
	switch t := datum.(type) {
	case parser.DInt:
		d = parser.DFloat(t)
	case parser.DFloat:
		d = t
	// case parser.DDecimal:
	// TODO(nvanbenschoten) add support for decimal variance and stddev
	// aggregation functions. Will require adding decimal.Sqrt() to library.
	default:
		return util.Errorf("unexpected VARIANCE argument type: %s", datum.Type())
	}

	// Uses the Knuth/Welford method for accurately computing variance online in a
	// single pass. See http://www.johndcook.com/blog/standard_deviation/ and
	// https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Online_algorithm.
	a.count++
	delta := d - a.mean
	a.mean += delta / parser.DFloat(a.count)
	a.sqrDiff += delta * (d - a.mean)
	return nil
}

func (a *sumAggregate) Add(datum parser.Datum) error {
	if datum == parser.DNull {
		return nil
	}
	if a.sum == nil {
		a.sum = datum
		return nil
	}

	switch t := datum.(type) {
	case parser.DInt:
		if v, ok := a.sum.(parser.DInt); ok {
			a.sum = v + t
			return nil
		}

	case parser.DFloat:
		if v, ok := a.sum.(parser.DFloat); ok {
			a.sum = v + t
			return nil
		}
	}

	return fmt.Errorf("unexpected SUM argument type: %s", datum.Type())
}

func (n *valuesNode) Less(i, j int) bool {
	// TODO(pmattis): An alternative to this type of field-based comparison would
	// be to construct a sort-key per row using encodeTableKey(). Using a
	// sort-key approach would likely fit better with a disk-based sort.
	ra, rb := n.rows[i], n.rows[j]
	for _, c := range n.ordering {
		var da, db parser.Datum
		if c.direction == encoding.Ascending {
			da = ra[c.colIdx]
			db = rb[c.colIdx]
		} else {
			da = rb[c.colIdx]
			db = ra[c.colIdx]
		}
		// TODO(pmattis): This is assuming that the datum types are compatible. I'm
		// not sure this always holds as `CASE` expressions can return different
		// types for a column for different rows. Investigate how other RDBMs
		// handle this.
		if c := da.Compare(db); c < 0 {
			return true
		} else if c > 0 {
			return false
		}
	}
	return true
}

// DecodeTableValue decodes a value encoded by EncodeTableValue.
func DecodeTableValue(a *DatumAlloc, valType parser.Datum, b []byte) (parser.Datum, []byte, error) {
	// TODO(dan): Merge this and DecodeTableKey.
	if len(b) == 0 {
		return nil, nil, util.Errorf("empty slice")
	}
	if roachpb.ValueType(b[0]) == roachpb.ValueType_NULL {
		return parser.DNull, b[1:], nil
	}
	var err error
	switch valType.(type) {
	case *parser.DBool:
		var i int64
		b, i, err = roachpb.DecodeIntValue(b)
		// No need to chunk allocate DBool as MakeDBool returns either
		// parser.DBoolTrue or parser.DBoolFalse.
		return parser.MakeDBool(parser.DBool(i != 0)), b, err
	case *parser.DInt:
		var i int64
		b, i, err = roachpb.DecodeIntValue(b)
		return a.NewDInt(parser.DInt(i)), b, err
	case *parser.DFloat:
		var f float64
		b, f, err = roachpb.DecodeFloatValue(b)
		return a.NewDFloat(parser.DFloat(f)), b, err
	case *parser.DDecimal:
		var d *inf.Dec
		b, d, err = roachpb.DecodeDecimalValue(b)
		dd := a.NewDDecimal(parser.DDecimal{})
		dd.Set(d)
		return dd, b, err
	case *parser.DString:
		var data []byte
		b, data, err = roachpb.DecodeBytesValue(b)
		return a.NewDString(parser.DString(data)), b, err
	case *parser.DBytes:
		var data []byte
		b, data, err = roachpb.DecodeBytesValue(b)
		return a.NewDBytes(parser.DBytes(data)), b, err
	case *parser.DDate:
		var i int64
		b, i, err = roachpb.DecodeIntValue(b)
		return a.NewDDate(parser.DDate(i)), b, err
	case *parser.DTimestamp:
		var t time.Time
		b, t, err = roachpb.DecodeTimeValue(b)
		return a.NewDTimestamp(parser.DTimestamp{Time: t}), b, err
	case *parser.DTimestampTZ:
		var t time.Time
		b, t, err = roachpb.DecodeTimeValue(b)
		return a.NewDTimestampTZ(parser.DTimestampTZ{Time: t}), b, err
	case *parser.DInterval:
		var d duration.Duration
		b, d, err = roachpb.DecodeDurationValue(b)
		return a.NewDInterval(parser.DInterval{Duration: d}), b, err
	default:
		return nil, nil, util.Errorf("TODO(pmattis): decoded index value: %s", valType.Type())
	}
}

// DecodeTableValue decodes a value encoded by EncodeTableValue.
func DecodeTableValue(a *DatumAlloc, valType parser.Datum, b []byte) (parser.Datum, []byte, error) {
	_, dataOffset, _, typ, err := encoding.DecodeValueTag(b)
	if err != nil {
		return nil, b, err
	}
	if typ == encoding.Null {
		return parser.DNull, b[dataOffset:], nil
	}
	switch valType.(type) {
	case *parser.DBool:
		var x bool
		b, x, err = encoding.DecodeBoolValue(b)
		// No need to chunk allocate DBool as MakeDBool returns either
		// parser.DBoolTrue or parser.DBoolFalse.
		return parser.MakeDBool(parser.DBool(x)), b, err
	case *parser.DInt:
		var i int64
		b, i, err = encoding.DecodeIntValue(b)
		return a.NewDInt(parser.DInt(i)), b, err
	case *parser.DFloat:
		var f float64
		b, f, err = encoding.DecodeFloatValue(b)
		return a.NewDFloat(parser.DFloat(f)), b, err
	case *parser.DDecimal:
		var d *inf.Dec
		b, d, err = encoding.DecodeDecimalValue(b)
		dd := a.NewDDecimal(parser.DDecimal{})
		dd.Set(d)
		return dd, b, err
	case *parser.DString:
		var data []byte
		b, data, err = encoding.DecodeBytesValue(b)
		return a.NewDString(parser.DString(data)), b, err
	case *parser.DBytes:
		var data []byte
		b, data, err = encoding.DecodeBytesValue(b)
		return a.NewDBytes(parser.DBytes(data)), b, err
	case *parser.DDate:
		var i int64
		b, i, err = encoding.DecodeIntValue(b)
		return a.NewDDate(parser.DDate(i)), b, err
	case *parser.DTimestamp:
		var t time.Time
		b, t, err = encoding.DecodeTimeValue(b)
		return a.NewDTimestamp(parser.DTimestamp{Time: t}), b, err
	case *parser.DTimestampTZ:
		var t time.Time
		b, t, err = encoding.DecodeTimeValue(b)
		return a.NewDTimestampTZ(parser.DTimestampTZ{Time: t}), b, err
	case *parser.DInterval:
		var d duration.Duration
		b, d, err = encoding.DecodeDurationValue(b)
		return a.NewDInterval(parser.DInterval{Duration: d}), b, err
	default:
		return nil, nil, errors.Errorf("TODO(pmattis): decoded index value: %s", valType.Type())
	}
}

// CheckColumnType verifies that a given value is compatible
// with the type requested by the column. If the value is a
// placeholder, the type of the placeholder gets populated.
func CheckColumnType(col ColumnDescriptor, val parser.Datum, args parser.MapArgs) error {
	if val == parser.DNull {
		return nil
	}

	var ok bool
	var err error
	var set parser.Datum
	switch col.Type.Kind {
	case ColumnType_BOOL:
		_, ok = val.(*parser.DBool)
		set, err = args.SetInferredType(val, parser.TypeBool)
	case ColumnType_INT:
		_, ok = val.(*parser.DInt)
		set, err = args.SetInferredType(val, parser.TypeInt)
	case ColumnType_FLOAT:
		_, ok = val.(*parser.DFloat)
		set, err = args.SetInferredType(val, parser.TypeFloat)
	case ColumnType_DECIMAL:
		_, ok = val.(*parser.DDecimal)
		set, err = args.SetInferredType(val, parser.TypeDecimal)
	case ColumnType_STRING:
		_, ok = val.(*parser.DString)
		set, err = args.SetInferredType(val, parser.TypeString)
	case ColumnType_BYTES:
		_, ok = val.(*parser.DBytes)
		if !ok {
			_, ok = val.(*parser.DString)
		}
		set, err = args.SetInferredType(val, parser.TypeBytes)
	case ColumnType_DATE:
		_, ok = val.(*parser.DDate)
		set, err = args.SetInferredType(val, parser.TypeDate)
	case ColumnType_TIMESTAMP:
		_, ok = val.(*parser.DTimestamp)
		set, err = args.SetInferredType(val, parser.TypeTimestamp)
	case ColumnType_TIMESTAMPTZ:
		_, ok = val.(*parser.DTimestampTZ)
		set, err = args.SetInferredType(val, parser.TypeTimestampTZ)
	case ColumnType_INTERVAL:
		_, ok = val.(*parser.DInterval)
		set, err = args.SetInferredType(val, parser.TypeInterval)
	default:
		return util.Errorf("unsupported column type: %s", col.Type.Kind)
	}
	// Check that the value cast has succeeded.
	// We ignore the case where it has failed because val was a DArg,
	// which is signalled by SetInferredType returning a non-nil assignment.
	if !ok && set == nil {
		return fmt.Errorf("value type %s doesn't match type %s of column %q",
			val.Type(), col.Type.Kind, col.Name)
	}
	return err
}

// SetDatum initializes the EncDatum with the given Datum.
func (ed *EncDatum) SetDatum(typ ColumnType_Kind, d parser.Datum) {
	if d == nil {
		panic("nil datum given")
	}
	if d != parser.DNull && !typ.ToDatumType().TypeEqual(d) {
		panic(fmt.Sprintf("invalid datum type given: %s, expected %s",
			d.Type(), typ.ToDatumType().Type()))
	}
	ed.Type = typ
	ed.encoded = nil
	ed.Datum = d
}

// SanitizeVarFreeExpr verifies a default expression is valid, has the
// correct type and contains no variable expressions.
func SanitizeVarFreeExpr(expr parser.Expr, expectedType parser.Datum, context string) error {
	if parser.ContainsVars(expr) {
		return exprContainsVarsError(context, expr)
	}
	typedExpr, err := parser.TypeCheck(expr, nil, expectedType)
	if err != nil {
		return err
	}
	if defaultType := typedExpr.ReturnType(); !expectedType.TypeEqual(defaultType) {
		return incompatibleExprTypeError(context, expectedType, defaultType)
	}
	return nil
}

// SanitizeDefaultExpr verifies a default expression is valid and has the
// correct type.
func SanitizeDefaultExpr(expr parser.Expr, colDatumType parser.Datum) error {
	typedExpr, err := parser.TypeCheck(expr, nil, colDatumType)
	if err != nil {
		return err
	}
	if defaultType := typedExpr.ReturnType(); !colDatumType.TypeEqual(defaultType) {
		return incompatibleColumnDefaultTypeError(colDatumType, defaultType)
	}
	if parser.ContainsVars(typedExpr) {
		return defaultContainsPlaceholdersError(typedExpr)
	}
	return nil
}

func protoFromDatum(datum parser.Datum) driver.Datum {
	if datum == parser.DNull {
		return driver.Datum{}
	}

	switch vt := datum.(type) {
	case parser.DBool:
		return driver.Datum{
			Payload: &driver.Datum_BoolVal{BoolVal: bool(vt)},
		}
	case parser.DInt:
		return driver.Datum{
			Payload: &driver.Datum_IntVal{IntVal: int64(vt)},
		}
	case parser.DFloat:
		return driver.Datum{
			Payload: &driver.Datum_FloatVal{FloatVal: float64(vt)},
		}
	case *parser.DDecimal:
		return driver.Datum{
			Payload: &driver.Datum_DecimalVal{DecimalVal: vt.Dec.String()},
		}
	case parser.DBytes:
		return driver.Datum{
			Payload: &driver.Datum_BytesVal{BytesVal: []byte(vt)},
		}
	case parser.DString:
		return driver.Datum{
			Payload: &driver.Datum_StringVal{StringVal: string(vt)},
		}
	case parser.DDate:
		return driver.Datum{
			Payload: &driver.Datum_DateVal{DateVal: int64(vt)},
		}
	case parser.DTimestamp:
		wireTimestamp := driver.Timestamp(vt.Time)
		return driver.Datum{
			Payload: &driver.Datum_TimeVal{
				TimeVal: &wireTimestamp,
			},
		}
	case parser.DInterval:
		return driver.Datum{
			Payload: &driver.Datum_IntervalVal{IntervalVal: vt.Nanoseconds()},
		}
	default:
		panic(util.Errorf("unsupported result type: %s", datum.Type()))
	}
}

func (n *scanNode) initWhere(where *parser.Where) *roachpb.Error {
	if where == nil {
		return nil
	}
	n.filter, n.pErr = n.resolveQNames(where.Expr)
	if n.pErr == nil {
		var whereType parser.Datum
		var err error
		whereType, err = n.filter.TypeCheck(n.planner.evalCtx.Args)
		n.pErr = roachpb.NewError(err)
		if n.pErr == nil {
			if !(whereType == parser.DummyBool || whereType == parser.DNull) {
				n.pErr = roachpb.NewUErrorf("argument of WHERE must be type %s, not type %s", parser.DummyBool.Type(), whereType.Type())
			}
		}
	}
	if n.pErr == nil {
		// Normalize the expression (this will also evaluate any branches that are
		// constant).
		var err error
		n.filter, err = n.planner.parser.NormalizeExpr(n.planner.evalCtx, n.filter)
		n.pErr = roachpb.NewError(err)
	}
	if n.pErr == nil {
		n.filter, n.pErr = n.planner.expandSubqueries(n.filter, 1)
	}
	return n.pErr
}

// marshalColumnValue returns a Go primitive value equivalent of val, of the
// type expected by col. If val's type is incompatible with col, or if
// col's type is not yet implemented, an error is returned.
func marshalColumnValue(col ColumnDescriptor, val parser.Datum) (interface{}, error) {
	if val == parser.DNull {
		return nil, nil
	}

	switch col.Type.Kind {
	case ColumnType_BOOL:
		if v, ok := val.(parser.DBool); ok {
			return bool(v), nil
		}
	case ColumnType_INT:
		if v, ok := val.(parser.DInt); ok {
			return int64(v), nil
		}
	case ColumnType_FLOAT:
		if v, ok := val.(parser.DFloat); ok {
			return float64(v), nil
		}
	case ColumnType_STRING:
		if v, ok := val.(parser.DString); ok {
			return string(v), nil
		}
	case ColumnType_BYTES:
		if v, ok := val.(parser.DBytes); ok {
			return string(v), nil
		}
	case ColumnType_DATE:
		if v, ok := val.(parser.DDate); ok {
			return v.Time, nil
		}
	case ColumnType_TIMESTAMP:
		if v, ok := val.(parser.DTimestamp); ok {
			return v.Time, nil
		}
	case ColumnType_INTERVAL:
		if v, ok := val.(parser.DInterval); ok {
			return v.Duration, nil
		}
		// case ColumnType_DECIMAL:
	default:
		return nil, fmt.Errorf("unsupported type: %s", val.Type())
	}
	return nil, fmt.Errorf("value type %s doesn't match type %s of column %q",
		val.Type(), col.Type.Kind, col.Name)
}

func TestEncDatumCompare(t *testing.T) {
	a := &DatumAlloc{}
	rng, _ := randutil.NewPseudoRand()

	for typ := ColumnType_Kind(0); int(typ) < len(ColumnType_Kind_value); typ++ {
		// Generate two datums d1 < d2
		var d1, d2 parser.Datum
		for {
			d1 = RandDatum(rng, typ, false)
			d2 = RandDatum(rng, typ, false)
			if cmp := d1.Compare(d2); cmp < 0 {
				break
			}
		}
		v1 := &EncDatum{}
		v1.SetDatum(typ, d1)
		v2 := &EncDatum{}
		v2.SetDatum(typ, d2)

		if val, err := v1.Compare(a, v2); err != nil {
			t.Fatal(err)
		} else if val != -1 {
			t.Errorf("compare(1, 2) = %d", val)
		}

		asc := DatumEncoding_ASCENDING_KEY
		desc := DatumEncoding_DESCENDING_KEY
		noncmp := DatumEncoding_VALUE

		checkEncDatumCmp(t, a, v1, v2, asc, asc, -1, false)
		checkEncDatumCmp(t, a, v2, v1, asc, asc, +1, false)
		checkEncDatumCmp(t, a, v1, v1, asc, asc, 0, false)
		checkEncDatumCmp(t, a, v2, v2, asc, asc, 0, false)

		checkEncDatumCmp(t, a, v1, v2, desc, desc, -1, false)
		checkEncDatumCmp(t, a, v2, v1, desc, desc, +1, false)
		checkEncDatumCmp(t, a, v1, v1, desc, desc, 0, false)
		checkEncDatumCmp(t, a, v2, v2, desc, desc, 0, false)

		checkEncDatumCmp(t, a, v1, v2, noncmp, noncmp, -1, true)
		checkEncDatumCmp(t, a, v2, v1, desc, noncmp, +1, true)
		checkEncDatumCmp(t, a, v1, v1, asc, desc, 0, true)
		checkEncDatumCmp(t, a, v2, v2, desc, asc, 0, true)
	}
}

func makeDriverDatum(datum parser.Datum) (driver.Datum, error) {
	if datum == parser.DNull {
		return driver.Datum{}, nil
	}

	switch vt := datum.(type) {
	case parser.DBool:
		return driver.Datum{
			Payload: &driver.Datum_BoolVal{BoolVal: bool(vt)},
		}, nil
	case parser.DInt:
		return driver.Datum{
			Payload: &driver.Datum_IntVal{IntVal: int64(vt)},
		}, nil
	case parser.DFloat:
		return driver.Datum{
			Payload: &driver.Datum_FloatVal{FloatVal: float64(vt)},
		}, nil
	case parser.DBytes:
		return driver.Datum{
			Payload: &driver.Datum_BytesVal{BytesVal: []byte(vt)},
		}, nil
	case parser.DString:
		return driver.Datum{
			Payload: &driver.Datum_StringVal{StringVal: string(vt)},
		}, nil
	case parser.DDate:
		return driver.Datum{
			Payload: &driver.Datum_DateVal{DateVal: int64(vt)},
		}, nil
	case parser.DTimestamp:
		wireTimestamp := driver.Timestamp(vt.Time)
		return driver.Datum{
			Payload: &driver.Datum_TimeVal{
				TimeVal: &wireTimestamp,
			},
		}, nil
	case parser.DInterval:
		return driver.Datum{
			Payload: &driver.Datum_IntervalVal{IntervalVal: vt.Nanoseconds()},
		}, nil
	default:
		return driver.Datum{}, fmt.Errorf("unsupported result type: %s", datum.Type())
	}
}

func checkResultDatum(datum parser.Datum) *roachpb.Error {
	if datum == parser.DNull {
		return nil
	}

	switch datum.(type) {
	case parser.DBool:
	case parser.DInt:
	case parser.DFloat:
	case parser.DBytes:
	case parser.DString:
	case parser.DDate:
	case parser.DTimestamp:
	case parser.DInterval:
	default:
		return roachpb.NewUErrorf("unsupported result type: %s", datum.Type())
	}
	return nil
}

func (a *sumAggregate) add(datum parser.Datum) error {
	if datum == parser.DNull {
		return nil
	}
	if a.sum == nil {
		switch t := datum.(type) {
		case *parser.DDecimal:
			// Make copy of decimal to allow for modification later.
			dd := &parser.DDecimal{}
			dd.Set(&t.Dec)
			datum = dd
		}
		a.sum = datum
		return nil
	}

	switch t := datum.(type) {
	case parser.DInt:
		if v, ok := a.sum.(parser.DInt); ok {
			a.sum = v + t
			return nil
		}

	case parser.DFloat:
		if v, ok := a.sum.(parser.DFloat); ok {
			a.sum = v + t
			return nil
		}

	case *parser.DDecimal:
		if v, ok := a.sum.(*parser.DDecimal); ok {
			v.Add(&v.Dec, &t.Dec)
			a.sum = v
			return nil
		}
	}

	return util.Errorf("unexpected SUM argument type: %s", datum.Type())
}