diff --git a/api.go b/api.go index d5a17b4a8..59f4105bb 100644 --- a/api.go +++ b/api.go @@ -1613,6 +1613,11 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu if err != nil { api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) } + } else if len(req.TimestampValues) > 0 { + err = field.importTimestampValue(qcx, req.ColumnIDs, req.TimestampValues, options) + if err != nil { + api.server.logger.Printf("import error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err) + } } else if len(req.FloatValues) > 0 { err = field.importFloatValue(qcx, req.ColumnIDs, req.FloatValues, options) if err != nil { diff --git a/api_test.go b/api_test.go index ad223db47..a4db231f6 100644 --- a/api_test.go +++ b/api_test.go @@ -446,8 +446,6 @@ func TestAPI_ImportValue(t *testing.T) { }) t.Run("ValTimestampField", func(t *testing.T) { - t.Skip("TODO(benbjohnson): timestamp") - ctx := context.Background() index := "valts" field := "fts" @@ -484,7 +482,7 @@ func TestAPI_ImportValue(t *testing.T) { } PanicOn(qcx.Finish()) - query := fmt.Sprintf("Row(%s>'1833-11-24T17:31:50Z')", field) + query := fmt.Sprintf("Row(%s>='1833-11-24T17:31:50Z')", field) // 6s after MinTimestamp // Query node0. if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: query}); err != nil { diff --git a/executor.go b/executor.go index 6ec17d73e..0fdc88d0f 100644 --- a/executor.go +++ b/executor.go @@ -1030,7 +1030,7 @@ func (e *executor) executeFieldValueCallShard(ctx context.Context, qcx *Qcx, fie other.FloatVal = 0 other.Val = 0 } else if field.Type() == FieldTypeTimestamp { - other.TimestampVal = time.Unix(0, value*int64(TimeUnitNano(field.Options().TimeUnit))) + other.TimestampVal = time.Unix(0, value*int64(TimeUnitNanos(field.Options().TimeUnit))) } return other, nil @@ -4492,7 +4492,6 @@ func (e *executor) executeExtractShard(ctx context.Context, qcx *Qcx, index stri } func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) { - span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowShard") defer span.Finish() @@ -4543,7 +4542,6 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, // Simply return row if times are not set. timeNotSet := fromTime.IsZero() && toTime.IsZero() if c.Name == "Row" && timeNotSet { - frag := e.Holder.fragment(index, fieldName, viewStandard, shard) if frag == nil { return NewRow(), nil @@ -4602,7 +4600,6 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string, // executeRowBSIGroupShard executes a range(bsiGroup) call for a local shard. func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) { - span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowBSIGroupShard") defer span.Finish() @@ -7400,7 +7397,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index case 0: return nil, nil case 1: - return time.Unix(0, int64(ids[0])*int64(TimeUnitNano(field.Options().TimeUnit))).UTC(), nil + return time.Unix(0, int64(ids[0])*int64(TimeUnitNanos(field.Options().TimeUnit))).UTC(), nil default: return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) } @@ -8268,7 +8265,7 @@ func getScaledInt(f *Field, v interface{}) (int64, error) { } else if opt.Type == FieldTypeTimestamp { switch tv := v.(type) { case time.Time: - value = tv.UnixNano() / TimeUnitNano(f.options.TimeUnit) + value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit) default: return 0, errors.Errorf("unexpected timestamp value type %T, val %v", tv, tv) } diff --git a/field.go b/field.go index 6666222c4..aebd93477 100644 --- a/field.go +++ b/field.go @@ -210,8 +210,8 @@ func OptFieldTypeInt(min, max int64) FieldOption { // provide any respective configuration values. func OptFieldTypeTimestamp(min, max time.Time, timeUnit string) FieldOption { return func(fo *FieldOptions) error { - minValue := min.UnixNano() / TimeUnitNano(timeUnit) - maxValue := max.UnixNano() / TimeUnitNano(timeUnit) + minValue := min.UnixNano() / TimeUnitNanos(timeUnit) + maxValue := max.UnixNano() / TimeUnitNanos(timeUnit) if fo.Type != "" { return errors.Errorf("field type is already set to: %s", fo.Type) } @@ -1421,7 +1421,7 @@ func (f *Field) MaxForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) dec := pql.NewDecimal(max+bsig.Base, bsig.Scale) valCount.DecimalVal = &dec } else if f.Options().Type == FieldTypeTimestamp { - valCount.TimestampVal = time.Unix(0, (max+bsig.Base)*TimeUnitNano(f.options.TimeUnit)).UTC() + valCount.TimestampVal = time.Unix(0, (max+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() } else { valCount.Val = max + bsig.Base } @@ -1467,7 +1467,7 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error) dec := pql.NewDecimal(min+bsig.Base, bsig.Scale) valCount.DecimalVal = &dec } else if f.Options().Type == FieldTypeTimestamp { - valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNano(f.options.TimeUnit)).UTC() + valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() } else { valCount.Val = min + bsig.Base } @@ -1602,6 +1602,19 @@ func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, return f.importValue(qcx, columnIDs, ivalues, options) } +func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, options *ImportOptions) error { + ivalues := make([]int64, len(values)) + bsig := f.bsiGroup(f.name) + if bsig == nil { + return errors.Wrap(ErrBSIGroupNotFound, f.name) + } + + for i, t := range values { + ivalues[i] = t.UnixNano() / TimeUnitNanos(f.options.TimeUnit) + } + return f.importValue(qcx, columnIDs, ivalues, options) +} + // importValue bulk imports range-encoded value data. func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, options *ImportOptions) (err0 error) { viewName := viewBSIGroupPrefix + f.name @@ -1921,10 +1934,10 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { ForeignIndex string `json:"foreignIndex"` }{ o.Type, - time.Unix(0, o.Base*TimeUnitNano(o.TimeUnit)).UTC(), + time.Unix(0, o.Base*TimeUnitNanos(o.TimeUnit)).UTC(), o.BitDepth, - time.Unix(0, o.Min.Value*TimeUnitNano(o.TimeUnit)).UTC(), - time.Unix(0, o.Max.Value*TimeUnitNano(o.TimeUnit)).UTC(), + time.Unix(0, o.Min.Value*TimeUnitNanos(o.TimeUnit)).UTC(), + time.Unix(0, o.Max.Value*TimeUnitNanos(o.TimeUnit)).UTC(), o.Keys, o.TimeUnit, o.ForeignIndex, @@ -1965,12 +1978,12 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { // MinTimestamp returns the minimum value for a timestamp field. func (o FieldOptions) MinTimestamp() time.Time { - return time.Unix(0, o.Min.ToInt64(0)*int64(TimeUnitNano(o.TimeUnit))) + return time.Unix(0, o.Min.ToInt64(0)*int64(TimeUnitNanos(o.TimeUnit))) } // MaxTimestamp returns the maxnimum value for a timestamp field. func (o FieldOptions) MaxTimestamp() time.Time { - return time.Unix(0, o.Max.ToInt64(0)*int64(TimeUnitNano(o.TimeUnit))) + return time.Unix(0, o.Max.ToInt64(0)*int64(TimeUnitNanos(o.TimeUnit))) } // List of bsiGroup types. @@ -2162,8 +2175,8 @@ func IsValidTimeUnit(unit string) bool { } } -// TimeUnitNano returns the number of nanoseconds in unit. -func TimeUnitNano(unit string) int64 { +// TimeUnitNanos returns the number of nanoseconds in unit. +func TimeUnitNanos(unit string) int64 { switch unit { case TimeUnitSeconds: return int64(time.Second) diff --git a/http/handler.go b/http/handler.go index 6dac8ff9e..0af8f5fa6 100644 --- a/http/handler.go +++ b/http/handler.go @@ -1342,16 +1342,16 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { fos = append(fos, pilosa.OptFieldTypeDecimal(scale, minmax...)) case pilosa.FieldTypeTimestamp: if req.Options.Min == nil { - min := pql.NewDecimal(pilosa.MinTimestamp.UnixNano()/pilosa.TimeUnitNano(*req.Options.TimeUnit), 0) + min := pql.NewDecimal(pilosa.MinTimestamp.UnixNano()/pilosa.TimeUnitNanos(*req.Options.TimeUnit), 0) req.Options.Min = &min } if req.Options.Max == nil { - max := pql.NewDecimal(pilosa.MaxTimestamp.UnixNano()/pilosa.TimeUnitNano(*req.Options.TimeUnit), 0) + max := pql.NewDecimal(pilosa.MaxTimestamp.UnixNano()/pilosa.TimeUnitNanos(*req.Options.TimeUnit), 0) req.Options.Max = &max } fos = append(fos, pilosa.OptFieldTypeTimestamp( - time.Unix(0, req.Options.Min.ToInt64(0)*pilosa.TimeUnitNano(*req.Options.TimeUnit)).UTC(), - time.Unix(0, req.Options.Max.ToInt64(0)*pilosa.TimeUnitNano(*req.Options.TimeUnit)).UTC(), + time.Unix(0, req.Options.Min.ToInt64(0)*pilosa.TimeUnitNanos(*req.Options.TimeUnit)).UTC(), + time.Unix(0, req.Options.Max.ToInt64(0)*pilosa.TimeUnitNanos(*req.Options.TimeUnit)).UTC(), *req.Options.TimeUnit, )) case pilosa.FieldTypeTime: