diff --git a/api.go b/api.go index ca4db1c9a..1fca0fa99 100644 --- a/api.go +++ b/api.go @@ -2006,11 +2006,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string return fmt.Errorf("adding decimal field to codec: %w", err) } case "timestamp": - // TODO: codec will likely need to be updated to reflect changes (increase) to timestamp range - // i.e. values are not all being converted to Nanos and being stored at the granularity - // specified by user and taking advantage of the range that provides. - nanos := TimeUnitNanos(field.options.TimeUnit) - if err = codec.AddTimestampField(field.name, time.Duration(nanos), field.options.Base); err != nil { + if err = codec.AddTimestampField(field.name, field.options.TimeUnit, field.options.Base); err != nil { return fmt.Errorf("adding timestamp field to codec: %w", err) } default: diff --git a/executor.go b/executor.go index d3eb8a07a..1e0b1c83d 100644 --- a/executor.go +++ b/executor.go @@ -1590,8 +1590,6 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str return nil, errors.Wrap(err, "translating value to timestamp") } results[i] = t.Format(time.RFC3339Nano) - - // results[i] = FormatTimestampNano(int64(val), bsig.Base, field.options.TimeUnit) } result = DistinctTimestamp{Name: fieldName, Values: results} return result, nil @@ -7919,7 +7917,7 @@ func (vc *ValCount) smaller(other ValCount) ValCount { return vc.decimalSmaller(other) } else if vc.FloatVal != 0 || other.FloatVal != 0 { return vc.floatSmaller(other) - } else if !vc.TimestampVal.Equal(time.Time{}) || !other.TimestampVal.Equal(time.Time{}) { + } else if !vc.TimestampVal.IsZero() || !other.TimestampVal.IsZero() { return vc.timestampSmaller(other) } if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) { @@ -7938,6 +7936,8 @@ func (vc *ValCount) smaller(other ValCount) ValCount { } } +// timestampSmaller returns the smaller of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) timestampSmaller(other ValCount) ValCount { if other.TimestampVal.Equal(time.Time{}) { return *vc @@ -7956,6 +7956,8 @@ func (vc *ValCount) timestampSmaller(other ValCount) ValCount { } } +// decimalSmaller returns the smaller of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) decimalSmaller(other ValCount) ValCount { if other.DecimalVal == nil { return *vc @@ -7973,6 +7975,8 @@ func (vc *ValCount) decimalSmaller(other ValCount) ValCount { } } +// floatSmaller returns the smaller of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) floatSmaller(other ValCount) ValCount { if vc.Count == 0 || (other.FloatVal < vc.FloatVal && other.Count > 0) { return other @@ -8012,6 +8016,8 @@ func (vc *ValCount) larger(other ValCount) ValCount { } } +// timestampLarger returns the larger of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) timestampLarger(other ValCount) ValCount { if other.TimestampVal.Equal(time.Time{}) { return *vc @@ -8030,6 +8036,8 @@ func (vc *ValCount) timestampLarger(other ValCount) ValCount { } } +// decimalLarger returns the larger of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) decimalLarger(other ValCount) ValCount { if other.DecimalVal == nil { return *vc @@ -8047,6 +8055,8 @@ func (vc *ValCount) decimalLarger(other ValCount) ValCount { } } +// floatLarger returns the larger of the two (vc or other), while merging the count +// if they are equal. func (vc *ValCount) floatLarger(other ValCount) ValCount { if vc.Count == 0 || (other.FloatVal > vc.FloatVal && other.Count > 0) { return other diff --git a/executor_test.go b/executor_test.go index f22c79c09..387a3ab34 100644 --- a/executor_test.go +++ b/executor_test.go @@ -7493,7 +7493,7 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { } } -func toCSV(s string) string { +func lineBreaker(s string) string { return strings.Join(strings.Split(s, " "), "\n") + "\n" } @@ -7553,44 +7553,44 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { // Rows { query: `Rows(f1, from='2019-08-04T14:36', to='2019-08-04T16:00')`, - csvVerifier: toCSV("R4 R5"), + csvVerifier: lineBreaker("R4 R5"), }, { query: `Rows(f1, from='2019-08-04T14', to='2019-08-04T17:00')`, - csvVerifier: toCSV("R4 R5 R6"), + csvVerifier: lineBreaker("R4 R5 R6"), }, { query: `Rows(f1, from='2019-08-04', to='2019-08-05')`, - csvVerifier: toCSV("R3 R4 R5 R6"), + csvVerifier: lineBreaker("R3 R4 R5 R6"), }, { query: `Rows(f1, from='2019-08', to='2019-12')`, - csvVerifier: toCSV("R2 R3 R4 R5 R6 R7"), + csvVerifier: lineBreaker("R2 R3 R4 R5 R6 R7"), }, { query: `Rows(f1, from='2019', to='2020')`, - csvVerifier: toCSV("R1 R2 R3 R4 R5 R6 R7 R8"), + csvVerifier: lineBreaker("R1 R2 R3 R4 R5 R6 R7 R8"), }, // Row { query: `Row(f2='R', from='2019-08-04T14:36', to='2019-08-04T16:00')`, - csvVerifier: toCSV("C4 C5"), + csvVerifier: lineBreaker("C4 C5"), }, { query: `Row(f2='R', from='2019-08-04T14', to='2019-08-04T17:00')`, - csvVerifier: toCSV("C4 C5 C6"), + csvVerifier: lineBreaker("C4 C5 C6"), }, { query: `Row(f2='R', from='2019-08-04', to='2019-08-05')`, - csvVerifier: toCSV("C3 C4 C5 C6"), + csvVerifier: lineBreaker("C3 C4 C5 C6"), }, { query: `Row(f2='R', from='2019-08', to='2019-12')`, - csvVerifier: toCSV("C2 C3 C4 C5 C6 C7"), + csvVerifier: lineBreaker("C2 C3 C4 C5 C6 C7"), }, { query: `Row(f2='R', from='2019', to='2020')`, - csvVerifier: toCSV("C1 C2 C3 C4 C5 C6 C7 C8"), + csvVerifier: lineBreaker("C1 C2 C3 C4 C5 C6 C7 C8"), }, } @@ -7862,39 +7862,39 @@ userB }, { query: `Row(unix_sec="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_sec="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Row(unix_milli="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_milli="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Row(unix_micro="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_micro="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Row(unix_nano="1833-11-24T17:31:44Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_nano="2106-02-07T06:28:16Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Union(Row(unix_nano="2106-02-07T06:28:16Z"), Row(unix_micro="0001-01-01T00:00:01Z"))`, - csvVerifier: toCSV("userA\nuserB"), + csvVerifier: lineBreaker("userA\nuserB"), }, { query: `Set("userA", unix_milli="2000-12-31T23:59:59.999Z")`, @@ -8155,59 +8155,59 @@ userC }, { query: `Row(unix_sec_min="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_sec_max="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userC"), + csvVerifier: lineBreaker("userC"), }, { query: `Row(unix_sec_max="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_milli_min="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_milli_min="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userC"), + csvVerifier: lineBreaker("userC"), }, { query: `Row(unix_micro_max="0001-01-01T00:00:01Z")`, - csvVerifier: toCSV("userC"), + csvVerifier: lineBreaker("userC"), }, { query: `Row(unix_micro_max="9999-12-31T23:59:59Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_nano_min="1833-11-24T17:31:44Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_nano_min="1969-12-31T23:59:59.999999999Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Row(unix_nano_min="1970-01-01T00:00:00Z")`, - csvVerifier: toCSV("userC"), + csvVerifier: lineBreaker("userC"), }, { query: `Row(unix_nano_max="2106-02-07T06:28:16Z")`, - csvVerifier: toCSV("userA"), + csvVerifier: lineBreaker("userA"), }, { query: `Row(unix_nano_max="1970-01-01T00:00:00.000000001Z")`, - csvVerifier: toCSV("userB"), + csvVerifier: lineBreaker("userB"), }, { query: `Row(unix_nano_max="1970-01-01T00:00:00Z")`, - csvVerifier: toCSV("userC"), + csvVerifier: lineBreaker("userC"), }, { query: `Union(Row(unix_nano_max="2106-02-07T06:28:16Z"), Row(unix_micro_max="0001-01-01T00:00:01Z"))`, - csvVerifier: toCSV("userA\nuserC"), + csvVerifier: lineBreaker("userA\nuserC"), }, { query: `Set("userA", unix_sec_min="2000-12-31T23:59:59.999Z")`, diff --git a/field.go b/field.go index 4e0606fef..3fc7e18e2 100644 --- a/field.go +++ b/field.go @@ -1833,7 +1833,7 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard // of timestamps are already relative to the epoch (base). // So a user may set an epoch to 2022-03-01 as the start of a race // and import finishing times in seconds. - // Timestamps ingested as timestamps are of coure absolute, but by the time + // Timestamps ingested as timestamps are of course absolute, but by the time // we get here it would be a relative integer. if f.Type() != FieldTypeTimestamp { min -= bsig.Base diff --git a/ingest/codec.go b/ingest/codec.go index c1e11f375..8c4abf5a8 100644 --- a/ingest/codec.go +++ b/ingest/codec.go @@ -44,7 +44,7 @@ type Codec interface { AddBoolField(name string) error AddIntField(name string, keys KeyTranslator) error AddDecimalField(name string, scale int64) error - AddTimestampField(name string, scale time.Duration, epoch int64) error + AddTimestampField(name string, scale string, epoch int64) error // Parse data from a reader into the vectors. // This must only be called once on a codec. @@ -70,14 +70,14 @@ type fieldCodec struct { currentOp *FieldOperation decode jsonDecFn encode jsonEncFn - // For timestamp: scale-in-nanoseconds; for instance, if scaleUnit is - // 1,000,000,000, we are storing numbers-of-seconds since the Unix epoch. - // The actual value recorded in BSI will be offset by the field's - // epoch, but we don't need to know that. // For decimal: Decimal digits of precision. So for instance, with // scale 2, scaleUnit is 100, "1" is stored as 100 and "1.2" is stored as // 120. scaleUnit int64 + // For timestamp: we use timeUnit to determine the scale at which + // to store a timestamp. For example if the timeUnit is milliseconds + // we store the number of milliseconds from the given epoch. + timeUnit string scale int64 epoch int64 // used only by Timestamp fields scratch []uint64 // reusable scratch space for sets of values @@ -282,10 +282,10 @@ func (codec *JSONCodec) AddBoolField(name string) error { // passed to this function should be the offset from the Unix epoch to the // desired epoch, in the same scale. (So if the scale is milliseconds, // it should be the Unix timestamp in seconds, times 1000.) -func (codec *JSONCodec) AddTimestampField(name string, timeScale time.Duration, epoch int64) error { +func (codec *JSONCodec) AddTimestampField(name string, timeScale string, epoch int64) error { fieldCodec := &fieldCodec{ fieldType: FieldTypeTimeStamp, - scaleUnit: int64(timeScale), + timeUnit: timeScale, epoch: epoch, } fieldCodec.decode = fieldCodec.DecodeTimeValue @@ -590,7 +590,7 @@ func (j *fieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.ValueType if err != nil { return fmt.Errorf("parsing timestamp: %w", err) } - j.currentOp.AddSignedPair(recID, (stamp.UnixNano()/j.scaleUnit)-j.epoch) + j.currentOp.AddSignedPair(recID, TimestampToVal(j.timeUnit, stamp)-j.epoch) case jsonparser.Number: // We could in theory convert this to a time, then convert it // back, by multiplying by scaleUnit, then dividing. Or... not. @@ -609,7 +609,11 @@ func (j *fieldCodec) EncodeTimeValue(dst *jsonBuffer, values []uint64, signed [] if len(signed) == 0 { return errors.New("encoding time value: no value provided") } - dst.EncodeTime(time.Unix(0, (signed[0]+j.epoch)*j.scaleUnit).UTC()) + t, err := ValToTimestamp(j.timeUnit, signed[0]+j.epoch) + if err != nil { + return errors.Wrap(err, "translating value to timestamp") + } + dst.EncodeTime(t) return nil } @@ -1132,3 +1136,35 @@ type errFieldNotFound struct { func (err errFieldNotFound) Error() string { return fmt.Sprintf("field not found: %q", err.field) } + +// TimestampToVal takes a time unit and a time.Time and converts it to an integer value +func TimestampToVal(unit string, ts time.Time) int64 { + switch unit { + case "s": + return ts.Unix() + case "ms": + return ts.UnixMilli() + case "us": + return ts.UnixMicro() + case "ns": + return ts.UnixNano() + } + return 0 + +} + +// ValToTimestamp takes a timeunit and an integer value and converts it to time.Time +func ValToTimestamp(unit string, val int64) (time.Time, error) { + switch unit { + case "s": + return time.Unix(val, 0).UTC(), nil + case "ms": + return time.UnixMilli(val).UTC(), nil + case "us", "μs": + return time.UnixMicro(val).UTC(), nil + case "ns": + return time.Unix(0, val).UTC(), nil + default: + return time.Time{}, errors.Errorf("Unknown time unit: '%v'", unit) + } +} diff --git a/ingest/codec_test.go b/ingest/codec_test.go index d36a0b757..838328981 100644 --- a/ingest/codec_test.go +++ b/ingest/codec_test.go @@ -58,7 +58,7 @@ func TestEncode(t *testing.T) { if err != nil { t.Fatalf("can't parse sample epoch time: %v", err) } - _ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000) _ = codec.AddDecimalField("dec", 2) _ = codec.AddBoolField("bool") @@ -74,7 +74,7 @@ func TestEncode(t *testing.T) { _ = codec.AddTimeQuantumField("tqkeys", newStableTranslator()) _ = codec.AddIntField("int", nil) _ = codec.AddIntField("intkeys", newStableTranslator()) - _ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000) _ = codec.AddDecimalField("dec", 2) _ = codec.AddBoolField("bool") @@ -291,7 +291,7 @@ func TestCodecErrors(t *testing.T) { if err != nil { t.Fatalf("can't parse sample epoch time: %v", err) } - _ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000) _ = codec.AddDecimalField("dec", 2) _ = codec.AddBoolField("bool") @@ -475,7 +475,7 @@ func TestSimpleCodec(t *testing.T) { if err != nil { t.Fatalf("can't parse sample epoch time: %v", err) } - _ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000) + _ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000) _ = codec.AddDecimalField("dec", 2) _ = codec.AddBoolField("bool") var nextShard = uint64(1<