diff --git a/sql3/planner/opbulkinsert.go b/sql3/planner/opbulkinsert.go index 5895271ff..d8bd33413 100644 --- a/sql3/planner/opbulkinsert.go +++ b/sql3/planner/opbulkinsert.go @@ -300,8 +300,9 @@ func (i *bulkInsertSourceCSVRowIter) Next(ctx context.Context) (types.Row, error } else { return nil, sql3.NewErrTypeConversionOnMap(0, 0, evalValue, mapColumn.colType.TypeDescription()) } + } else { + result[idx] = time.UnixMilli(intVal).UTC() } - result[idx] = time.UnixMilli(intVal).UTC() case *parser.DataTypeString: result[idx] = evalValue diff --git a/sql3/planner/oppqlaggregate.go b/sql3/planner/oppqlaggregate.go index d1c85daf8..15c7ef7ae 100644 --- a/sql3/planner/oppqlaggregate.go +++ b/sql3/planner/oppqlaggregate.go @@ -263,20 +263,36 @@ func (i *pqlAggregateRowIter) Next(ctx context.Context) (types.Row, error) { i.resultValue = int64(actualResult) case pilosa.ValCount: - if actualResult.DecimalVal == nil { + switch t := i.aggregate.Type().(type) { + case *parser.DataTypeInt: + i.resultValue = int64(actualResult.Val) + + case *parser.DataTypeDecimal: if i.aggregate.AggType() == types.AGGREGATE_AVG { - average := float64(actualResult.Val) / float64(actualResult.Count) - i.resultValue = pql.NewDecimal(int64(average*10000), 4) - } else { - i.resultValue = int64(actualResult.Val) - } - } else { - if i.aggregate.AggType() == types.AGGREGATE_AVG { - average := actualResult.DecimalVal.Float64() / float64(actualResult.Count) - i.resultValue = pql.NewDecimal(int64(average*10000), 4) + if actualResult.DecimalVal == nil { + average := float64(actualResult.Val) / float64(actualResult.Count) + daverage, err := pql.FromFloat64WithScale(average, int(t.Scale)) + if err != nil { + return nil, err + } + i.resultValue = daverage + } else { + average := actualResult.DecimalVal.Float64() / float64(actualResult.Count) + daverage, err := pql.FromFloat64WithScale(average, int(t.Scale)) + if err != nil { + return nil, err + } + i.resultValue = daverage + } } else { i.resultValue = *actualResult.DecimalVal } + + case *parser.DataTypeTimestamp: + i.resultValue = actualResult.TimestampVal + + default: + return nil, sql3.NewErrInternalf("unhandled return type '%T'", i.aggregate.Type()) } default: return nil, sql3.NewErrInternalf("unexpected result type '%T'", queryResponse.Results[0]) diff --git a/sql3/sql_test.go b/sql3/sql_test.go index 053971c6a..4f45fd940 100644 --- a/sql3/sql_test.go +++ b/sql3/sql_test.go @@ -62,6 +62,8 @@ func TestSQL_Execute(t *testing.T) { m[headers[i].Name] = i } + // TODO(pok) - this will become increasingly problematic as result column headers + // are not unique and can be empty // Put the expRows in the same column order as the headers returned // by the query. exp := make([][]interface{}, len(sqltest.ExpRows)) diff --git a/sql3/test/defs/defs.go b/sql3/test/defs/defs.go index 41619d43c..b6b37ad17 100644 --- a/sql3/test/defs/defs.go +++ b/sql3/test/defs/defs.go @@ -167,6 +167,10 @@ var TableTests []TableTest = []TableTest{ joinTestsOrders, joinTests, + // bulk insert + bulkInsertTable, + bulkInsert, + //bool (batch logic) boolTests, @@ -181,3 +185,11 @@ func knownTimestamp() time.Time { } return tm } + +func timestampFromString(s string) time.Time { + tm, err := time.ParseInLocation(time.RFC3339, s, time.UTC) + if err != nil { + panic(err.Error()) + } + return tm +} diff --git a/sql3/test/defs/defs_aggregate.go b/sql3/test/defs/defs_aggregate.go index 611a53954..f77bc6194 100644 --- a/sql3/test/defs/defs_aggregate.go +++ b/sql3/test/defs/defs_aggregate.go @@ -427,14 +427,15 @@ var minmaxTests = TableTest{ srcHdr("i1", fldTypeInt, "min 0", "max 1000"), srcHdr("d1", fldTypeDecimal2), srcHdr("s1", fldTypeString), + srcHdr("ts1", fldTypeTimestamp), ), srcRows( - srcRow(int64(1), int64(10), float64(10), string("foo")), - srcRow(int64(2), int64(10), float64(10), string("foo")), - srcRow(int64(3), int64(11), float64(11), string("foo")), - srcRow(int64(4), int64(12), float64(12), string("foo")), - srcRow(int64(5), int64(12), float64(12), string("foo")), - srcRow(int64(6), int64(13), float64(13), string("foo")), + srcRow(int64(1), int64(10), float64(10), string("foo"), timestampFromString("2013-07-15T01:18:46Z")), + srcRow(int64(2), int64(10), float64(10), string("foo"), timestampFromString("2014-07-15T01:18:46Z")), + srcRow(int64(3), int64(11), float64(11), string("foo"), timestampFromString("2015-07-15T01:18:46Z")), + srcRow(int64(4), int64(12), float64(12), string("foo"), timestampFromString("2016-07-15T01:18:46Z")), + srcRow(int64(5), int64(12), float64(12), string("foo"), timestampFromString("2017-07-15T01:18:46Z")), + srcRow(int64(6), int64(13), float64(13), string("foo"), timestampFromString("2018-07-15T01:18:46Z")), ), ), SQLTests: []SQLTest{ @@ -521,5 +522,18 @@ var minmaxTests = TableTest{ ), Compare: CompareExactUnordered, }, + { + SQLs: sqls( + "select min(ts1) as min_val, max(ts1) as max_val from minmax_test", + ), + ExpHdrs: hdrs( + hdr("min_val", fldTypeTimestamp), + hdr("max_val", fldTypeTimestamp), + ), + ExpRows: rows( + row(timestampFromString("2013-07-15T01:18:46Z"), timestampFromString("2018-07-15T01:18:46Z")), + ), + Compare: CompareExactUnordered, + }, }, } diff --git a/sql3/test/defs/defs_bulkinsert.go b/sql3/test/defs/defs_bulkinsert.go new file mode 100644 index 000000000..b07b300f8 --- /dev/null +++ b/sql3/test/defs/defs_bulkinsert.go @@ -0,0 +1,59 @@ +package defs + +// join tests +var bulkInsertTable = TableTest{ + name: "bulkInsertTable", + Table: tbl( + "bulktest", + srcHdrs( + srcHdr("_id", fldTypeString), + srcHdr("id_col", fldTypeID), + srcHdr("string_col", fldTypeString), + srcHdr("int_col", fldTypeInt), + srcHdr("decimal_col", fldTypeDecimal2), + srcHdr("bool_col", fldTypeBool), + srcHdr("time_col", fldTypeTimestamp), + srcHdr("stringset_col", fldTypeStringSet), + srcHdr("idset_col", fldTypeIDSet), + ), + ), + SQLTests: nil, +} + +var bulkInsert = TableTest{ + name: "bulkInsert", + SQLTests: []SQLTest{ + { + name: "timestamp-csv-text", + SQLs: sqls( + `BULK INSERT INTO + bulktest (_id, id_col, string_col, int_col,decimal_col, bool_col, time_col, stringset_col, idset_col) + map (0 ID, 1 STRING, 2 INT, 3 DECIMAL(2), 4 BOOL, 5 TIMESTAMP, 6 STRINGSET, 7 IDSET) + transform(@1, @0, @1, @2, @3, @4, @5, @6, @7) + FROM x'1,TEST,-123,1.12,0,2013-07-15T01:18:46Z,stringset1, 1 + 2,TEST2,321,31.2,1,2014-07-15T01:18:46Z,stringset1, 1 + 1,TEST,-123,1.12,0,2013-07-15T01:18:46Z,stringset2, 2' + with + BATCHSIZE 10000 + format 'CSV' + input 'STREAM';`, + ), + ExpHdrs: hdrs(), + ExpRows: rows(), + Compare: CompareExactOrdered, + }, + { + name: "leftjoin", + SQLs: sqls( + "select time_col from bulktest where _id = 'TEST2';", + ), + ExpHdrs: hdrs( + hdr("time_col", fldTypeTimestamp), + ), + ExpRows: rows( + row(timestampFromString("2014-07-15T01:18:46Z")), + ), + Compare: CompareExactOrdered, + }, + }, +}