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

update codec to reflect changes to timstamp range

Browse source 
also address a few minor feedback comments from review
This commit is contained in:
Samir Patel 2022-07-15 13:01:15 -05:00 • committed by Samir Patel
parent 33fce8a05b
commit bd5dc760b7
6 changed files with 98 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api.go
View file

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

16
executor.go
View file

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

68
executor_test.go
View file

@ -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")`,

2
field.go
View file

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

54
ingest/codec.go
View file

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

8
ingest/codec_test.go
View file

@ -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<<shardwidth.Exponent) + 5