scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix timestamp value import

Browse source
This commit is contained in:
Ben Johnson 2021-04-09 10:55:12 -06:00
parent 8802856120
commit 5defbe3ef2
5 changed files with 37 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
api.go
View file

@ -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 {

4
api_test.go
View file

@ -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 {

9
executor.go
View file

@ -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)
}

35
field.go
View file

@ -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)

8
http/handler.go
View file

@ -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: