Merge branch 'master' of https://github.com/kshvakov/clickhouse

Author: kshvakov

lib/column/column.go

@@ -137,6 +137,8 @@ func Factory(name, chType string, timezone *time.Location) (Column, error) {
 		return parseFixedString(name, chType)
 	case strings.HasPrefix(chType, "Enum8"), strings.HasPrefix(chType, "Enum16"):
 		return parseEnum(name, chType)
+	case strings.HasPrefix(chType, "Decimal"):
+		return parseDecimal(name, chType)
 	}
 	return nil, fmt.Errorf("column: unhandled type %v", chType)
 }

lib/column/decimal.go

@@ -0,0 +1,188 @@
+package column
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/kshvakov/clickhouse/lib/binary"
+)
+
+// Table of powers of 10 for fast casting from floating types to decimal type
+// representations.
+var factors10 = []float64{
+	1e0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8, 1e9, 1e10, 1e11, 1e12, 1e13,
+	1e14, 1e15, 1e16, 1e17, 1e18,
+}
+
+// Decimal represents Decimal(P, S) ClickHouse. Since there is support for
+// int128 in Golang, the implementation does not support to 128-bits decimals
+// as well. Decimal is represented as integral. Also floating-point types are
+// supported for query parameters.
+type Decimal struct {
+	base
+	nobits    int // its domain is {32, 64}
+	precision int
+	scale     int
+}
+
+func (d *Decimal) Read(decoder *binary.Decoder) (interface{}, error) {
+	switch d.nobits {
+	case 32:
+		return decoder.Int32()
+	case 64:
+		return decoder.Int64()
+	default:
+		return nil, errors.New("unachievable execution path")
+	}
+}
+
+func (d *Decimal) Write(encoder *binary.Encoder, v interface{}) error {
+	switch d.nobits {
+	case 32:
+		return d.write32(encoder, v)
+	case 64:
+		return d.write64(encoder, v)
+	default:
+		return errors.New("unachievable execution path")
+	}
+}
+
+func (d *Decimal) float2int32(floating float64) int32 {
+	fixed := int32(floating * factors10[d.scale])
+	return fixed
+}
+
+func (d *Decimal) float2int64(floating float64) int64 {
+	fixed := int64(floating * factors10[d.scale])
+	return fixed
+}
+
+func (d *Decimal) write32(encoder *binary.Encoder, v interface{}) error {
+	switch v := v.(type) {
+	case int8:
+		return encoder.Int32(int32(v))
+	case int16:
+		return encoder.Int32(int32(v))
+	case int32:
+		return encoder.Int32(int32(v))
+	case int64:
+		return errors.New("narrowing type conversion from int64 to int32")
+
+	case uint8:
+		return encoder.Int32(int32(v))
+	case uint16:
+		return encoder.Int32(int32(v))
+	case uint32:
+		return errors.New("narrowing type conversion from uint32 to int32")
+	case uint64:
+		return errors.New("narrowing type conversion from uint64 to int32")
+
+	case float32:
+		fixed := d.float2int32(float64(v))
+		return encoder.Int32(fixed)
+	case float64:
+		fixed := d.float2int32(float64(v))
+		return encoder.Int32(fixed)
+	}
+
+	return &ErrUnexpectedType{
+		T:      v,
+		Column: d,
+	}
+	return errors.New("narrowing type conversion from int64 to int32")
+}
+
+func (d *Decimal) write64(encoder *binary.Encoder, v interface{}) error {
+	switch v := v.(type) {
+	case int:
+		return encoder.Int64(int64(v))
+	case int8:
+		return encoder.Int64(int64(v))
+	case int16:
+		return encoder.Int64(int64(v))
+	case int32:
+		return encoder.Int64(int64(v))
+	case int64:
+		return encoder.Int64(int64(v))
+
+	case uint8:
+		return encoder.Int64(int64(v))
+	case uint16:
+		return encoder.Int64(int64(v))
+	case uint32:
+		return encoder.Int64(int64(v))
+	case uint64:
+		return errors.New("narrowing type conversion from uint64 to int64")
+
+	case float32:
+		fixed := d.float2int64(float64(v))
+		return encoder.Int64(fixed)
+	case float64:
+		fixed := d.float2int64(float64(v))
+		return encoder.Int64(fixed)
+	}
+
+	return &ErrUnexpectedType{
+		T:      v,
+		Column: d,
+	}
+}
+
+func parseDecimal(name, chType string) (Column, error) {
+	switch {
+	case len(chType) < 12:
+		fallthrough
+	case !strings.HasPrefix(chType, "Decimal"):
+		fallthrough
+	case chType[7] != '(':
+		fallthrough
+	case chType[len(chType)-1] != ')':
+		return nil, fmt.Errorf("invalid Decimal format: '%s'", chType)
+	}
+
+	var params = strings.Split(chType[8:len(chType)-1], ",")
+
+	if len(params) != 2 {
+		return nil, fmt.Errorf("invalid Decimal format: '%s'", chType)
+	}
+
+	params[0] = strings.TrimSpace(params[0])
+	params[1] = strings.TrimSpace(params[1])
+
+	var err error
+	var decimal = &Decimal{
+		base: base{
+			name:   name,
+			chType: chType,
+		},
+	}
+
+	if decimal.precision, err = strconv.Atoi(params[0]); err != nil {
+		return nil, fmt.Errorf("'%s' is not Decimal type: %s", chType, err)
+	} else if decimal.precision < 1 {
+		return nil, errors.New("wrong precision of Decimal type")
+	}
+
+	if decimal.scale, err = strconv.Atoi(params[1]); err != nil {
+		return nil, fmt.Errorf("'%s' is not Decimal type: %s", chType, err)
+	} else if decimal.scale < 0 || decimal.scale > decimal.precision {
+		return nil, errors.New("wrong scale of Decimal type")
+	}
+
+	switch {
+	case decimal.precision <= 9:
+		decimal.nobits = 32
+		decimal.valueOf = baseTypes[int32(0)]
+	case decimal.precision <= 18:
+		decimal.nobits = 64
+		decimal.valueOf = baseTypes[int64(0)]
+	case decimal.precision <= 38:
+		return nil, errors.New("Decimal128 is not supported")
+	default:
+		return nil, errors.New("precision of Decimal exceeds max bound")
+	}
+
+	return decimal, nil
+}