Handle TimeStamp columns by converting int64 to time.Time and vice versa

crate.go

@@ -10,13 +10,50 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"time"
 )
 
 // Crate conn structure
 type CrateDriver struct {
 	Url string // Crate http endpoint url
 }
 
+//GeoPoint represents Crate GeoPoint column
+type GeoPoint struct {
+	Lat float64
+	Lon float64
+}
+
+//CrateArray represents an Array column type
+type CrateArray []interface{}
+
+//Scan : Implements Scanner interface to populate a GeoPoint when the result is an array of 2 floats
+func (gp *GeoPoint) Scan(src interface{}) error {
+	if b, ok := src.([]interface{}) ; ok && len(b) == 2 {
+		var err error
+		if gp.Lon, err = b[0].(json.Number).Float64(); err == nil {
+			if gp.Lat, err = b[1].(json.Number).Float64(); err == nil {
+				return nil
+			}
+		}
+		return fmt.Errorf("failed to convert %v to GeoPoint : %v", src, err)
+	}
+	return fmt.Errorf("failed to convert %v to GeoPoint", src)
+}
+
+//Scan : Implements Scanner interface to populate a CrateArray from the incoming data
+func (arr *CrateArray) Scan(src interface{}) error {
+	if srcArr, ok := src.([]interface{}) ; ok  {
+		*arr = make ([]interface{}, len(srcArr))
+		for  i, obj:=range(srcArr) {
+			(*arr)[i] = obj
+		}
+		return nil
+	}
+	return fmt.Errorf("failed to convert %v to CrateArray", src)
+}
+
+
 // Init a new "Connection" to a Crate Data Storage instance.
 // Note that the connection is not tested until the first query.
 func (c *CrateDriver) Open(crate_url string) (driver.Conn, error) {
@@ -53,6 +90,39 @@ type endpointQuery struct {
 	Args []driver.Value `json:"args,omitempty"`
 }
 
+//CheckNamedValue Convert map, CrateArray, time & GeoPoint arguments to DB format.
+func (c *CrateDriver) CheckNamedValue(v *driver.NamedValue) error {
+	if obj, ok := v.Value.(map[string]interface{}) ; ok {
+		if len(obj) == 0 {
+			v.Value = "{}"
+		} else {
+			v.Value = obj
+		}
+		return nil
+	} else if ts, ok := v.Value.(time.Time) ; ok {
+		if ts.IsZero() {
+			v.Value = 0
+		} else {
+			v.Value = ts.In(time.UTC).UnixNano() / 1000000
+		}
+		return nil
+	} else if gp, ok := v.Value.(GeoPoint) ; ok {
+		//fmt.Printf("CheckNamedValue for Geopoint (%f,%f) \n", gp.Lon, gp.Lat)
+		nGp := make([]float64, 2)
+		nGp[0] = gp.Lon
+		nGp[1] = gp.Lat
+		v.Value = &nGp
+		return nil
+	} else if arr, ok := v.Value.(CrateArray) ; ok {
+		v.Value = arr
+		return nil
+	}
+	/*else {
+		fmt.Printf("CheckNamedValue  for %v -> %v\n", v.Name, v.Value)
+	}*/
+	return driver.ErrSkip
+}
+
 // Query the database using prepared statements.
 // Read: https://crate.io/docs/stable/sql/rest.html for more information about the returned response.
 // Example: crate.Query("SELECT * FROM sys.cluster LIMIT ?", 10)
@@ -65,7 +135,7 @@ func (c *CrateDriver) query(stmt string, args []driver.Value) (*endpointResponse
 		Stmt: stmt,
 	}
 
-	if len(args) > 0 {
+	if l:=len(args); l > 0 {
 		query.Args = args
 	}
 
@@ -74,7 +144,6 @@ func (c *CrateDriver) query(stmt string, args []driver.Value) (*endpointResponse
 	if err != nil {
 		return nil, err
 	}
-
 	data := bytes.NewReader(buf)
 
 	resp, err := http.Post(endpoint, "application/json", data)
@@ -123,8 +192,18 @@ func (c *CrateDriver) Query(stmt string, args []driver.Value) (driver.Rows, erro
 		columns:  res.Cols,
 		values:   res.Rows,
 		rowcount: res.Rowcount,
+		isSpecial:   make([]int64, len(res.Cols)),
+	}
+	tcount := len(res.ColumnTypes)
+	for i:=0; i<tcount; i++ {
+		if n, ok := res.ColumnTypes[i].(json.Number); ok {
+			if t, err := n.Int64(); err == nil {
+				if t == typeTimestamp || t == typeGeoPoint {
+					rows.isSpecial[i] = t
+				}
+			}
+		}
 	}
-
 	return rows, nil
 }
 
@@ -141,6 +220,7 @@ func (c *CrateDriver) Exec(stmt string, args []driver.Value) (result driver.Resu
 	return result, nil
 }
 
+
 // Result interface
 type Result struct {
 	affectedRows int64
@@ -161,6 +241,7 @@ func (r *Result) RowsAffected() (int64, error) {
 type Rows struct {
 	columns  []string
 	values   [][]interface{}
+	isSpecial   []int64	//Flags columns to convert to time.Time (type 11)
 	rowcount int64
 	pos      int64 // index position on the values array
 }
@@ -175,9 +256,37 @@ func (r *Rows) Next(dest []driver.Value) error {
 	if r.pos >= r.rowcount {
 		return io.EOF
 	}
-
 	for i := range dest {
-		dest[i] = r.values[r.pos][i]
+		if ((r.isSpecial[i] != 0) && (r.values[r.pos][i] != nil)) {
+			if r.isSpecial[i] == typeTimestamp {
+				if val, ok := r.values[r.pos][i].(json.Number); ok {
+					v, _ := val.Int64()
+					sec := v / int64(1000)
+					dest[i] = time.Unix(sec, (v-sec*int64(1000))*int64(1000000))
+				} else {
+					return fmt.Errorf("failed to convert column %s=%T to time", r.columns[i], r.values[r.pos][i])
+				}
+			} else if r.isSpecial[i] == typeGeoPoint {
+				if psrc, ok := r.values[r.pos][i].([]interface{}) ; ok && (len(psrc) == 2) {
+					var p GeoPoint
+					var err error
+					if p.Lon, err = psrc[0].(json.Number).Float64(); err != nil {
+						return fmt.Errorf("failed to convert to latitude %v", psrc[0])
+					}
+					if p.Lat, err = psrc[1].(json.Number).Float64(); err != nil {
+						return fmt.Errorf("failed to convert to longitude elem %v", psrc[1])
+					}
+					dest[i] = &p
+				} else if len(psrc) == 0 {
+					dest[i] = GeoPoint{}
+				} else {
+					return fmt.Errorf("failed to convert %v to GeoPoint", r.values[r.pos][i])
+				}
+			}
+		} else {
+			dest[i] = r.values[r.pos][i]
+			//fmt.Printf("Processing column %s : %+v / %s\n", r.columns[i], r.values[r.pos][i], reflect.TypeOf(r.values[r.pos][i]))
+		}
 	}
 
 	r.pos++