diff --git a/.gitlab/.gitlab-ci.yml b/.gitlab/.gitlab-ci.yml index bd70bf523..b2e84e7a8 100644 --- a/.gitlab/.gitlab-ci.yml +++ b/.gitlab/.gitlab-ci.yml @@ -364,10 +364,10 @@ upload to sonarcloud: rules: - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"' script: - - sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=coverage*.out,results/coverage*out -Dsonar.go.tests.reportPaths=test-report*.out -Dsonar.javascript.lcov.reportPaths=lattice/coverage/lcov.info + - sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=coverage*.out,results/coverage*out,idk/testdata/coverage*out -Dsonar.go.tests.reportPaths=test-report*.out,idk/testdata/report*out -Dsonar.javascript.lcov.reportPaths=lattice/coverage/lcov.info - cd ./idk - ls ./testdata - - sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=testdata/*coverage.out -Dsonar.go.tests.reportPaths=testdata/*report.out -Dsonar.coverage.exclusions=**/*_test.go -Dsonar.cpd.exclusions=**/*_test.go + - sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=testdata/*coverage.out,idk/testdata/coverage*out -Dsonar.go.tests.reportPaths=testdata/*report.out,idk/testdata/report*out -Dsonar.coverage.exclusions=**/*_test.go -Dsonar.cpd.exclusions=**/*_test.go needs: - job: run go tests future plg - job: run go tests future diff --git a/api.go b/api.go index b7ac24d69..22f5239cc 100644 --- a/api.go +++ b/api.go @@ -2005,8 +2005,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string return fmt.Errorf("adding decimal field to codec: %w", err) } case "timestamp": - 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/client/client.go b/client/client.go index a195f1a75..a508f9723 100644 --- a/client/client.go +++ b/client/client.go @@ -1701,6 +1701,7 @@ type SchemaOptions struct { TrackExistence bool `json:"trackExistence"` TimeUnit string `json:"timeUnit"` Base int64 `json:"base"` + Epoch time.Time `json:"epoch"` } func (so SchemaOptions) asIndexOptions() *IndexOptions { @@ -1726,6 +1727,7 @@ func (so SchemaOptions) asFieldOptions() *FieldOptions { noStandardView: so.NoStandardView, timeUnit: so.TimeUnit, base: so.Base, + epoch: so.Epoch, } } diff --git a/executor.go b/executor.go index 97d336b32..92e55f5a1 100644 --- a/executor.go +++ b/executor.go @@ -977,7 +977,11 @@ 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(TimeUnitNanos(field.Options().TimeUnit))) + ts, err := ValToTimestamp(field.Options().TimeUnit, value) + if err != nil { + return ValCount{}, err + } + other.TimestampVal = ts } return other, nil @@ -1614,7 +1618,11 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str cols := r.Pos.Columns() results := make([]string, len(cols)) for i, val := range cols { - results[i] = FormatTimestampNano(int64(val), bsig.Base, field.options.TimeUnit) + t, err := ValToTimestamp(field.options.TimeUnit, int64(val)+bsig.Base) + if err != nil { + return nil, errors.Wrap(err, "translating value to timestamp") + } + results[i] = t.Format(time.RFC3339Nano) } result = DistinctTimestamp{Name: fieldName, Values: results} return result, nil @@ -1655,6 +1663,11 @@ func (d DistinctTimestamp) ToRows(callback func(*proto.RowResponse) error) error return nil } +// ToTable implements the ToTabler interface for DistinctTimestamp +func (d DistinctTimestamp) ToTable() (*proto.TableResponse, error) { + return proto.RowsToTable(&d, len(d.Values)) +} + // Union returns the union of the values of `d` and `other` func (d *DistinctTimestamp) Union(other DistinctTimestamp) DistinctTimestamp { both := map[string]struct{}{} @@ -3222,13 +3235,16 @@ func (fr *FieldRow) Clone() (clone *FieldRow) { func (fr FieldRow) MarshalJSON() ([]byte, error) { if fr.Value != nil { if fr.FieldOptions.Type == FieldTypeTimestamp { - ts := FormatTimestampNano(int64(*fr.Value), fr.FieldOptions.Base, fr.FieldOptions.TimeUnit) + ts, err := ValToTimestamp(fr.FieldOptions.TimeUnit, int64(*fr.Value)+fr.FieldOptions.Base) + if err != nil { + return nil, errors.Wrap(err, "translating value to timestamp") + } return json.Marshal(struct { Field string `json:"field"` Value string `json:"value"` }{ Field: fr.Field, - Value: ts, + Value: ts.Format(time.RFC3339Nano), }) } else { return json.Marshal(struct { @@ -7610,12 +7626,17 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index } case FieldTypeTimestamp: datatype = "timestamp" + unit := field.Options().TimeUnit mapper = func(ids []uint64) (_ interface{}, err error) { switch len(ids) { case 0: return nil, nil case 1: - return time.Unix(0, int64(ids[0])*int64(TimeUnitNanos(field.Options().TimeUnit))).UTC(), nil + ts, err := ValToTimestamp(unit, int64(ids[0])) + if err != nil { + return nil, err + } + return ts, nil default: return nil, errors.Errorf("BSI field %q has too many values: %v", field.Name(), ids) } @@ -7676,6 +7697,38 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index return result, nil } +// 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 TimeUnitSeconds: + return time.Unix(val, 0).UTC(), nil + case TimeUnitMilliseconds: + return time.UnixMilli(val).UTC(), nil + case TimeUnitMicroseconds, TimeUnitUSeconds: + return time.UnixMicro(val).UTC(), nil + case TimeUnitNanoseconds: + return time.Unix(0, val).UTC(), nil + default: + return time.Time{}, errors.Errorf("Unknown time unit: '%v'", unit) + } +} + +// 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 TimeUnitSeconds: + return ts.Unix() + case TimeUnitMilliseconds: + return ts.UnixMilli() + case TimeUnitMicroseconds, TimeUnitUSeconds: + return ts.UnixMicro() + case TimeUnitNanoseconds: + return ts.UnixNano() + } + return 0 + +} + // detectRangeCall returns true if the call or one of its children contains a Range call // TODO: Remove at version 2.0 func (e *executor) detectRangeCall(c *pql.Call) bool { @@ -7982,6 +8035,8 @@ 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.IsZero() || !other.TimestampVal.IsZero() { + return vc.timestampSmaller(other) } if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) { return other @@ -7999,6 +8054,28 @@ 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 + } + if vc.Count == 0 || vc.TimestampVal.Equal(time.Time{}) || (other.TimestampVal.Before(vc.TimestampVal) && other.Count > 0) { + return other + } + extra := int64(0) + if vc.TimestampVal.Equal(other.TimestampVal) { + extra += other.Count + } + return ValCount{ + Val: vc.Val, + TimestampVal: vc.TimestampVal, + Count: vc.Count + extra, + } +} + +// 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 @@ -8016,6 +8093,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 @@ -8036,6 +8115,8 @@ func (vc *ValCount) larger(other ValCount) ValCount { return vc.decimalLarger(other) } else if vc.FloatVal != 0 || other.FloatVal != 0 { return vc.floatLarger(other) + } else if !vc.TimestampVal.Equal(time.Time{}) || !other.TimestampVal.Equal(time.Time{}) { + return vc.timestampLarger(other) } if vc.Count == 0 || (other.Val > vc.Val && other.Count > 0) { return other @@ -8053,6 +8134,28 @@ 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 + } + if vc.Count == 0 || vc.TimestampVal.Equal(time.Time{}) || (other.TimestampVal.After(vc.TimestampVal) && other.Count > 0) { + return other + } + extra := int64(0) + if vc.TimestampVal.Equal(other.TimestampVal) { + extra += other.Count + } + return ValCount{ + Val: vc.Val, + TimestampVal: vc.TimestampVal, + Count: vc.Count + extra, + } +} + +// 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 @@ -8070,6 +8173,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 @@ -8487,7 +8592,8 @@ func getScaledInt(f *Field, v interface{}) (int64, error) { } else if opt.Type == FieldTypeTimestamp { switch tv := v.(type) { case time.Time: - value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit) + v := TimestampToVal(f.options.TimeUnit, tv) + value = v case int64: value = tv default: diff --git a/executor_test.go b/executor_test.go index e641d8ef8..76904ce61 100644 --- a/executor_test.go +++ b/executor_test.go @@ -7245,6 +7245,8 @@ func TestVariousQueries(t *testing.T) { variousQueriesOnPercentiles(t, c) variousQueriesCountDistinctTimestamp(t, c) variousQueriesOnIntFields(t, c) + variousQueriesOnTimestampFields(t, c) + variousQueriesOnLargeEpoch(t, c) backupTest(t, c, "") // test backup/restore of all indexes }) } @@ -7495,6 +7497,10 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { } } +func lineBreaker(s string) string { + return strings.Join(strings.Split(s, " "), "\n") + "\n" +} + // tests for abbreviating time values in queries func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { ts := func(t time.Time) int64 { @@ -7541,10 +7547,6 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { return strings.Join(ss, "\n") + "\n" } - toCSV := func(s string) string { - return strings.Join(strings.Split(s, " "), "\n") + "\n" - } - type testCase struct { query string qrVerifier func(t *testing.T, resp pilosa.QueryResponse) @@ -7555,44 +7557,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"), }, } @@ -7705,6 +7707,598 @@ userG,-1,10,10,10 } } +// Constants used in TimestampField testing +var ( + minTime = pilosa.MinTimestamp + maxTime = pilosa.MaxTimestamp + minSec = minTime.Unix() + maxSec = maxTime.Unix() + minMilli = minTime.UnixMilli() + maxMilli = maxTime.UnixMilli() + minMicro = minTime.UnixMicro() + maxMicro = maxTime.UnixMicro() + minNano = pilosa.MinTimestampNano.UnixNano() + maxNano = pilosa.MaxTimestampNano.UnixNano() +) + +// variousQueriesOnTimestampFields tests queries on Timestamp Fields at various granularities using the default epoch +func variousQueriesOnTimestampFields(t *testing.T, c *test.Cluster) { + index := "ts_test01" + + // Testing whether the max and min timestamps can be represented for seconds + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_sec", pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s")) + c.ImportIntKey(t, index, "unix_sec", []test.IntKey{ + {Val: minSec, Key: "userA"}, + {Val: maxSec, Key: "userB"}, + }) + + // Testing whether the max and min timestamps can be represented for milliseconds + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_milli", pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "ms")) + c.ImportIntKey(t, index, "unix_milli", []test.IntKey{ + {Val: minMilli, Key: "userA"}, + {Val: maxMilli, Key: "userB"}, + }) + + // Testing whether the max and min timestamps can be represented for microseconds + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_micro", pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "us")) + c.ImportIntKey(t, index, "unix_micro", []test.IntKey{ + {Val: minMicro, Key: "userA"}, + {Val: maxMicro, Key: "userB"}, + }) + + // Note that min and max values that can be represented for Nanos is a much smaller range than any of the above granularities. + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_nano", pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "ns")) + c.ImportIntKey(t, index, "unix_nano", []test.IntKey{ + {Val: minNano, Key: "userA"}, + {Val: maxNano, Key: "userB"}, + }) + + splitSortBackToCSV := func(csvStr string) string { + if len(csvStr) == 0 { + return "" + } + ss := strings.Split(csvStr[:len(csvStr)-1], "\n") + sort.Strings(ss) + return strings.Join(ss, "\n") + "\n" + } + + type testCase struct { + query string + qrVerifier func(t *testing.T, resp pilosa.QueryResponse) + csvVerifier string + } + + tests := []testCase{ + { + query: "extract(All(), Rows(unix_sec))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_milli))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_micro))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_nano))", + csvVerifier: `userA,1833-11-24T17:31:44Z +userB,2106-02-07T06:28:16Z +`, + }, + { + query: "All()", + csvVerifier: `userA +userB +`, + }, + { + query: "count(All())", + csvVerifier: `2 +`, + }, + // Found an existing bug: Distinct on timestamp does not return values prior to the epoch. + // These tests need to be uncommented when issue is fixed. + // { + // query: "Distinct(field=unix_sec)", + // csvVerifier: `0001-01-01T00:00:01Z + // 9999-12-31T23:59:59Z + // `, + // }, + // { + // query: "Distinct(field=unix_milli)", + // csvVerifier: `0001-01-01T00:00:01Z + // 9999-12-31T23:59:59Z + // `, + // }, + { + query: "Min(unix_sec)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_sec)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_milli)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_milli)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_micro)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_micro)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_nano)", + csvVerifier: `1833-11-24T17:31:44Z,1 +`, + }, + { + query: "Max(unix_nano)", + csvVerifier: `2106-02-07T06:28:16Z,1 +`, + }, + { + query: "GroupBy(Rows(unix_micro))", + csvVerifier: `-62135596799000000,1 +253402300799000000,1 +`, + }, + { + query: `Row(unix_sec="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_sec="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Row(unix_milli="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_milli="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Row(unix_micro="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_micro="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Row(unix_nano="1833-11-24T17:31:44Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_nano="2106-02-07T06:28:16Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Union(Row(unix_nano="2106-02-07T06:28:16Z"), Row(unix_micro="0001-01-01T00:00:01Z"))`, + csvVerifier: lineBreaker("userA\nuserB"), + }, + { + query: `Set("userA", unix_milli="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + { + query: `Clear("userA", unix_milli="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + // Not Supported for Timestamp + // { + // query: `ClearRow(unix_milli="2000-12-31T23:59:59.999Z")`, + // csvVerifier: `true`, + // }, + } + + for i, tst := range tests { + t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { + tr := c.QueryGRPC(t, index, tst.query) + csvString, err := tableResponseToCSVString(tr) + if err != nil { + t.Fatal(err) + } + // verify everything after header + got := splitSortBackToCSV(csvString[strings.Index(csvString, "\n")+1:]) + if got != tst.csvVerifier { + t.Errorf("expected:\n%s\ngot:\n%s", tst.csvVerifier, got) + } + }) + } +} + +// variousQueriesOnLargeEpoch tests queries on TimestampFields when the epoch is set either to +// the min or max timestamp allowed. +func variousQueriesOnLargeEpoch(t *testing.T, c *test.Cluster) { + index := "ts_epoch_test20321" + // mag := int64(10000000000) + + // These large constants are close to min and max int64 but not quite since go has to account for the difference between Unix Epoch and Go's Epoch + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_sec_min", pilosa.OptFieldTypeTimestamp(minTime, "s")) + c.ImportIntKey(t, index, "unix_sec_min", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -minSec, Key: "userB"}, + {Val: -minSec + maxSec, Key: "userC"}, + }) + + // vprint.VV("-MaxSec: %+v", -maxSec) + // vprint.VV("-MaxSec + minsec: %+v", -maxSec+minSec) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_sec_max", pilosa.OptFieldTypeTimestamp(maxTime, "s")) + c.ImportIntKey(t, index, "unix_sec_max", []test.IntKey{ + {Val: 0, Key: "userA"}, // 9999-12-31 + // {Val: 1, Key: "userE"}, + // {Val: -1, Key: "userD"}, + {Val: -maxSec, Key: "userB"}, //1970.... + {Val: -maxSec + minSec, Key: "userC"}, //0001 + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_milli_min", pilosa.OptFieldTypeTimestamp(minTime, "ms")) + c.ImportIntKey(t, index, "unix_milli_min", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -minMilli, Key: "userB"}, + {Val: -minMilli + maxMilli, Key: "userC"}, + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_milli_max", pilosa.OptFieldTypeTimestamp(maxTime, "ms")) + c.ImportIntKey(t, index, "unix_milli_max", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -maxMilli, Key: "userB"}, + {Val: -maxMilli + minMilli, Key: "userC"}, + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_micro_min", pilosa.OptFieldTypeTimestamp(minTime, "us")) + c.ImportIntKey(t, index, "unix_micro_min", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -minMicro, Key: "userB"}, + {Val: -minMicro + maxMicro, Key: "userC"}, + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_micro_max", pilosa.OptFieldTypeTimestamp(maxTime, "us")) + c.ImportIntKey(t, index, "unix_micro_max", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -maxMicro, Key: "userB"}, + {Val: -maxMicro + minMicro, Key: "userC"}, + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_nano_min", pilosa.OptFieldTypeTimestamp(pilosa.MinTimestampNano, "ns")) + c.ImportIntKey(t, index, "unix_nano_min", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -minNano - 1, Key: "userB"}, + {Val: -minNano, Key: "userC"}, + }) + + c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_nano_max", pilosa.OptFieldTypeTimestamp(pilosa.MaxTimestampNano, "ns")) + c.ImportIntKey(t, index, "unix_nano_max", []test.IntKey{ + {Val: 0, Key: "userA"}, + {Val: -maxNano + 1, Key: "userB"}, + {Val: -maxNano, Key: "userC"}, + }) + + splitSortBackToCSV := func(csvStr string) string { + if csvStr == "" { + return "" + } + ss := strings.Split(csvStr[:len(csvStr)-1], "\n") + sort.Strings(ss) + return strings.Join(ss, "\n") + "\n" + } + + type testCase struct { + query string + qrVerifier func(t *testing.T, resp pilosa.QueryResponse) + csvVerifier string + } + + tests := []testCase{ + { + query: "extract(All(), Rows(unix_sec_min))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,1970-01-01T00:00:00Z +userC,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_sec_max))", + csvVerifier: `userA,9999-12-31T23:59:59Z +userB,1970-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z +`, + }, + { + query: "extract(All(), Rows(unix_milli_min))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,1970-01-01T00:00:00Z +userC,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_milli_max))", + csvVerifier: `userA,9999-12-31T23:59:59Z +userB,1970-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z +`, + }, + { + query: "extract(All(), Rows(unix_micro_min))", + csvVerifier: `userA,0001-01-01T00:00:01Z +userB,1970-01-01T00:00:00Z +userC,9999-12-31T23:59:59Z +`, + }, + { + query: "extract(All(), Rows(unix_micro_max))", + csvVerifier: `userA,9999-12-31T23:59:59Z +userB,1970-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z +`, + }, + { + query: "extract(All(), Rows(unix_nano_min))", + csvVerifier: `userA,1833-11-24T17:31:44Z +userB,1969-12-31T23:59:59.999999999Z +userC,1970-01-01T00:00:00Z +`, + }, + { + query: "extract(All(), Rows(unix_nano_max))", + csvVerifier: `userA,2106-02-07T06:28:16Z +userB,1970-01-01T00:00:00.000000001Z +userC,1970-01-01T00:00:00Z +`, + }, + { + query: "All()", + csvVerifier: `userA +userB +userC +`, + }, + { + query: "count(All())", + csvVerifier: `3 +`, + }, + { + query: "Min(unix_sec_min)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_sec_min)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_sec_max)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_sec_max)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_milli_min)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_milli_min)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Max(unix_milli_max)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_micro_min)", + csvVerifier: `0001-01-01T00:00:01Z,1 +`, + }, + { + query: "Max(unix_micro_max)", + csvVerifier: `9999-12-31T23:59:59Z,1 +`, + }, + { + query: "Min(unix_nano_min)", + csvVerifier: `1833-11-24T17:31:44Z,1 +`, + }, + { + query: "Max(unix_nano_min)", + csvVerifier: `1970-01-01T00:00:00Z,1 +`, + }, + { + query: "Min(unix_nano_max)", + csvVerifier: `1970-01-01T00:00:00Z,1 +`, + }, + { + query: "Max(unix_nano_max)", + csvVerifier: `2106-02-07T06:28:16Z,1 +`, + }, + { + query: "GroupBy(Rows(unix_micro_min))", + csvVerifier: `-62135596799000000,1 +0,1 +253402300799000000,1 +`, + }, + { + query: `Row(unix_sec_min="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_sec_max="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userC"), + }, + { + query: `Row(unix_sec_max="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_milli_min="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_milli_min="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userC"), + }, + { + query: `Row(unix_micro_max="0001-01-01T00:00:01Z")`, + csvVerifier: lineBreaker("userC"), + }, + { + query: `Row(unix_micro_max="9999-12-31T23:59:59Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_nano_min="1833-11-24T17:31:44Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_nano_min="1969-12-31T23:59:59.999999999Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Row(unix_nano_min="1970-01-01T00:00:00Z")`, + csvVerifier: lineBreaker("userC"), + }, + { + query: `Row(unix_nano_max="2106-02-07T06:28:16Z")`, + csvVerifier: lineBreaker("userA"), + }, + { + query: `Row(unix_nano_max="1970-01-01T00:00:00.000000001Z")`, + csvVerifier: lineBreaker("userB"), + }, + { + query: `Row(unix_nano_max="1970-01-01T00:00:00Z")`, + csvVerifier: lineBreaker("userC"), + }, + { + query: `Union(Row(unix_nano_max="2106-02-07T06:28:16Z"), Row(unix_micro_max="0001-01-01T00:00:01Z"))`, + csvVerifier: lineBreaker("userA\nuserC"), + }, + { + query: `Set("userA", unix_sec_min="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + { + query: "extract(All(), Rows(unix_sec_min))", + csvVerifier: `userA,2000-12-31T23:59:59Z +userB,1970-01-01T00:00:00Z +userC,9999-12-31T23:59:59Z +`, + }, + { + query: `Clear("userA", unix_sec_min="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + { + query: `Set("userA", unix_milli_max="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + { + query: "extract(All(), Rows(unix_milli_max))", + csvVerifier: `userA,2000-12-31T23:59:59.999Z +userB,1970-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z +`, + }, + { + query: `Clear("userA", unix_milli_max="2000-12-31T23:59:59.999Z")`, + csvVerifier: `true +`, + }, + { + query: `Set("userA", unix_nano_max="2050-02-01T00:00:00.000000002Z")`, + csvVerifier: `true +`, + }, + { + query: "extract(All(), Rows(unix_nano_max))", + csvVerifier: `userA,2050-02-01T00:00:00.000000002Z +userB,1970-01-01T00:00:00.000000001Z +userC,1970-01-01T00:00:00Z +`, + }, + { + query: `Clear("userA", unix_nano_max="1970-01-01T00:00:00.000000002Z")`, + csvVerifier: `true +`, + }, + { + query: `Set("userA", unix_nano_min="1969-12-31T23:59:59.999999998Z")`, + csvVerifier: `true +`, + }, + { + query: "extract(All(), Rows(unix_nano_min))", + csvVerifier: `userA,1969-12-31T23:59:59.999999998Z +userB,1969-12-31T23:59:59.999999999Z +userC,1970-01-01T00:00:00Z +`, + }, + { + query: `Clear("userA", unix_nano_min="1969-12-31T23:59:59.999999998Z")`, + csvVerifier: `true +`, + }, + } + + for i, tst := range tests { + t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { + tr := c.QueryGRPC(t, index, tst.query) + csvString, err := tableResponseToCSVString(tr) + if err != nil { + t.Fatal(err) + } + // verify everything after header + got := splitSortBackToCSV(csvString[strings.Index(csvString, "\n")+1:]) + if got != tst.csvVerifier { + t.Errorf("expected:\n%s\ngot:\n%s", tst.csvVerifier, got) + } + }) + } +} + var usersIndex = "users" func populateTestData(t *testing.T, c *test.Cluster) { @@ -8330,6 +8924,9 @@ func tableResponseToCSV(m *proto.TableResponse, w io.Writer) error { record = append(record, fmt.Sprintf("%v", col.GetBoolVal())) case "int64": record = append(record, fmt.Sprintf("%v", col.GetInt64Val())) + case "timestamp": + record = append(record, fmt.Sprintf("%v", col.GetTimestampVal())) + } } err := writer.Write(record) diff --git a/field.go b/field.go index 6a0762d58..e9f67c5d0 100644 --- a/field.go +++ b/field.go @@ -195,26 +195,59 @@ func OptFieldTypeInt(min, max int64) FieldOption { // provide any respective configuration values. func OptFieldTypeTimestamp(epoch time.Time, timeUnit string) FieldOption { return func(fo *FieldOptions) error { - // Check if the epoch will overflow when converted to nano. - if err := CheckUnixNanoOverflow(epoch); err != nil { - return err - } - epochValue := epoch.UnixNano() / TimeUnitNanos(timeUnit) if fo.Type != "" { return errors.Errorf("field type is already set to: %s", fo.Type) } - if timeUnit == "" { - return errors.Errorf("time unit required for timestamp field") - } else if !IsValidTimeUnit(timeUnit) { - return errors.Errorf("invalid time unit: %q", fo.TimeUnit) + + minTime := MinTimestamp + maxTime := MaxTimestamp + + var base, minInt, maxInt int64 + switch timeUnit { + case TimeUnitSeconds: + base = epoch.Unix() + minInt = minTime.Unix() - base + maxInt = maxTime.Unix() - base + case TimeUnitMilliseconds: + base = epoch.UnixMilli() + minInt = minTime.UnixMilli() - base + maxInt = maxTime.UnixMilli() - base + case TimeUnitMicroseconds, TimeUnitUSeconds: + base = epoch.UnixMicro() + minInt = minTime.UnixMicro() - base + maxInt = maxTime.UnixMicro() - base + case TimeUnitNanoseconds: + // Note: For nano, the min and max values are also the min and max integer + // values we support. Also, keep in mind that MinNano is a negative + // number. So if base is positive and we do MinNano - base...it would increase minInt + // beyond what we support. This isn't an issue with larger granularities. + base = epoch.UnixNano() + if base > 0 { + maxInt = MaxTimestampNano.UnixNano() - base + minInt = MinTimestampNano.UnixNano() + } else { + maxInt = MaxTimestampNano.UnixNano() + minInt = MinTimestampNano.UnixNano() - base + } + minTime = MinTimestampNano + maxTime = MaxTimestampNano + default: + return errors.Errorf("invalid time unit: '%q'", fo.TimeUnit) } + + if err := CheckEpochOutOfRange(epoch, minTime, maxTime); err != nil { + return err + } + fo.Type = FieldTypeTimestamp fo.TimeUnit = timeUnit - fo.Min = pql.NewDecimal(MinTimestamp.UnixNano()/TimeUnitNanos(timeUnit), 0) - fo.Max = pql.NewDecimal(MaxTimestamp.UnixNano()/TimeUnitNanos(timeUnit), 0) - fo.Base = epochValue + fo.Base = base + fo.Min = pql.NewDecimal(minInt, 0) + fo.Max = pql.NewDecimal(maxInt, 0) + return nil } + } // OptFieldTypeDecimal is a functional option for creating a `decimal` field. @@ -1416,14 +1449,19 @@ func (f *Field) SetValue(tx Tx, columnID uint64, value int64) (changed bool, err bsig := f.bsiGroup(f.name) if bsig == nil { return false, ErrBSIGroupNotFound - } else if value < bsig.Min { - return false, errors.Wrapf(ErrBSIGroupValueTooLow, "index = %v, field = %v, column ID = %v, value %v is smaller than min allowed %v", f.index, f.name, columnID, value, bsig.Min) - } else if value > bsig.Max { - return false, errors.Wrapf(ErrBSIGroupValueTooHigh, "index = %v, field = %v, column ID = %v, value %v is larger than max allowed %v", f.index, f.name, columnID, value, bsig.Max) } // Determine base value to store. baseValue := int64(value - bsig.Base) + //Timestamp expects incoming value to already be relative to epoch + if f.Type() == FieldTypeTimestamp { + value = baseValue + } + if value < bsig.Min { + return false, errors.Wrapf(ErrBSIGroupValueTooLow, "index = %v, field = %v, column ID = %v, value %v is smaller than min allowed %v", f.index, f.name, columnID, value, bsig.Min) + } else if value > bsig.Max { + return false, errors.Wrapf(ErrBSIGroupValueTooHigh, "index = %v, field = %v, column ID = %v, value %v is larger than max allowed %v", f.index, f.name, columnID, value, bsig.Max) + } requiredBitDepth := bitDepthInt64(baseValue) @@ -1565,8 +1603,14 @@ func (f *Field) valCountize(val int64, cnt uint64, bsig *bsiGroup) (ValCount, er dec := pql.NewDecimal(val+bsig.Base, bsig.Scale) valCount.DecimalVal = &dec } else if f.Options().Type == FieldTypeTimestamp { - valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() + ts, err := ValToTimestamp(f.options.TimeUnit, val+bsig.Base) + if err != nil { + return ValCount{}, errors.Wrap(err, "translating value to timestamp") + } + valCount.TimestampVal = ts + // valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC() } + valCount.Val = val + bsig.Base return valCount, nil } @@ -1753,7 +1797,7 @@ func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time } for i, t := range values { - ivalues[i] = t.UnixNano() / TimeUnitNanos(f.options.TimeUnit) + ivalues[i] = TimestampToVal(f.options.TimeUnit, t) } return f.importValue(qcx, columnIDs, ivalues, shard, options) } @@ -1796,23 +1840,23 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard if value < min { min = value } - if f.Type() == FieldTypeTimestamp { - scale := (TimeUnitNanos(f.options.TimeUnit)) - offset := f.options.Base * scale - dur := value * scale - if offset > 0 { - if dur > math.MaxInt64-offset { - return errors.Wrap(ErrBSIGroupValueTooHigh, "value + epoch is too far from Unix epoch") - } - } else if dur < math.MinInt64-offset { - return errors.Wrap(ErrBSIGroupValueTooLow, "value + epoch is too far from Unix epoch") - } - } + + } + + // Timestamps differ from other BSI fields in that integer representations + // 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 course absolute, but by the time + // we get here it would be a relative integer. + if f.Type() != FieldTypeTimestamp { + min -= bsig.Base + max -= bsig.Base } // Determine the highest bit depth required by the min & max. - requiredDepth := bitDepthInt64(min - bsig.Base) - if v := bitDepthInt64(max - bsig.Base); v > requiredDepth { + requiredDepth := bitDepthInt64(min) + if v := bitDepthInt64(max); v > requiredDepth { requiredDepth = v } // Increase bit depth if required. @@ -2078,6 +2122,11 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { o.Keys, }) case FieldTypeTimestamp: + epoch, err := ValToTimestamp(o.TimeUnit, o.Base) + if err != nil { + return nil, errors.Wrap(err, "translating val to timestamp") + } + return json.Marshal(struct { Type string `json:"type"` Epoch time.Time `json:"epoch"` @@ -2087,7 +2136,7 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { TimeUnit string `json:"timeUnit"` }{ o.Type, - time.Unix(0, o.Base*TimeUnitNanos(o.TimeUnit)).UTC(), + epoch, o.BitDepth, o.Min, o.Max, @@ -2129,16 +2178,6 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) { return nil, errors.Errorf("invalid field type: '%s'", o.Type) } -// MinTimestamp returns the minimum value for a timestamp field. -func (o FieldOptions) MinTimestamp() time.Time { - 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(TimeUnitNanos(o.TimeUnit))) -} - // List of bsiGroup types. const ( bsiGroupTypeInt = "int" @@ -2304,15 +2343,16 @@ func (f *Field) persistView(ctx context.Context, cvm *CreateViewMessage) error { return f.schemator.CreateView(ctx, cvm.Index, cvm.Field, cvm.View) } -// Timestamp field range. +// Timestamp field ranges. var ( - DefaultEpoch = time.Unix(0, 0).UTC() // 1970-01-01T00:00:00Z - - MinTimestamp = time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z - MaxTimestamp = time.Unix(1<<32, 0).UTC() // 2106-02-07T06:28:16Z + DefaultEpoch = time.Unix(0, 0).UTC() // 1970-01-01T00:00:00Z + MinTimestampNano = time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z + MaxTimestampNano = time.Unix(1<<32, 0).UTC() // 2106-02-07T06:28:16Z + MinTimestamp = time.Unix(-62135596799, 0).UTC() // 0001-01-01T00:00:01Z + MaxTimestamp = time.Unix(253402300799, 0).UTC() // 9999-12-31T23:59:59Z ) -// List of time units. +// Constants related to timestamp. const ( TimeUnitSeconds = "s" TimeUnitMilliseconds = "ms" @@ -2345,12 +2385,9 @@ func TimeUnitNanos(unit string) int64 { } } -func CheckUnixNanoOverflow(epoch time.Time) error { - if time.Unix(0, 0).After(epoch) { - if epoch.UnixNano() > 0 { - return errors.Errorf("custom epoch too far from Unix epoch: %s", epoch) - } - } else if epoch.UnixNano() < 0 { +// CheckEpochOutOfRange checks if the epoch is after max or before min +func CheckEpochOutOfRange(epoch, min, max time.Time) error { + if epoch.After(max) || epoch.Before(min) { return errors.Errorf("custom epoch too far from Unix epoch: %s", epoch) } return nil diff --git a/field_test.go b/field_test.go index 1c602416a..85299d4ea 100644 --- a/field_test.go +++ b/field_test.go @@ -307,6 +307,8 @@ func TestFieldInfoMarshal(t *testing.T) { } func TestCheckUnixNanoOverflow(t *testing.T) { + minNano = pilosa.MinTimestampNano.UnixNano() + maxNano = pilosa.MaxTimestampNano.UnixNano() tests := []struct { name string epoch time.Time @@ -314,28 +316,28 @@ func TestCheckUnixNanoOverflow(t *testing.T) { }{ { name: "too small", - epoch: time.Unix(-1, math.MinInt64), + epoch: time.Unix(-1, minNano), wantErr: true, }, { name: "just right-1", - epoch: time.Unix(0, math.MinInt64), + epoch: time.Unix(0, minNano), wantErr: false, }, { name: "just right-2", - epoch: time.Unix(0, math.MaxInt64), + epoch: time.Unix(0, maxNano), wantErr: false, }, { name: "too large", - epoch: time.Unix(1, math.MaxInt64), + epoch: time.Unix(1, maxNano), wantErr: true, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if err := pilosa.CheckUnixNanoOverflow(tt.epoch); (err != nil) != tt.wantErr { + if err := pilosa.CheckEpochOutOfRange(tt.epoch, pilosa.MinTimestampNano, pilosa.MaxTimestampNano); (err != nil) != tt.wantErr { t.Errorf("checkUnixNanoOverflow() error = %v, wantErr %v", err, tt.wantErr) } }) diff --git a/fragment.go b/fragment.go index f4cdd0b7f..6e5891121 100644 --- a/fragment.go +++ b/fragment.go @@ -828,6 +828,7 @@ func (f *fragment) min(tx Tx, filter *Row, bitDepth uint64) (min int64, count ui // minUnsigned the lowest value without considering the sign bit. Filter is required. func (f *fragment) minUnsigned(tx Tx, filter *Row, bitDepth uint64) (min int64, count uint64, err error) { + count = filter.Count() for i := int(bitDepth - 1); i >= 0; i-- { row, err := f.row(tx, uint64(bsiOffsetBit+i)) if err != nil { @@ -879,6 +880,7 @@ func (f *fragment) max(tx Tx, filter *Row, bitDepth uint64) (max int64, count ui // maxUnsigned the highest value without considering the sign bit. Filter is required. func (f *fragment) maxUnsigned(tx Tx, filter *Row, bitDepth uint64) (max int64, count uint64, err error) { + count = filter.Count() for i := int(bitDepth - 1); i >= 0; i-- { row, err := f.row(tx, uint64(bsiOffsetBit+i)) if err != nil { diff --git a/idk/csv/csv_test.go b/idk/csv/csv_test.go index e6af1c30a..17f8198b5 100644 --- a/idk/csv/csv_test.go +++ b/idk/csv/csv_test.go @@ -310,8 +310,8 @@ func testDirTree(tree string) (string, error) { func TestVariousOORValues(t *testing.T) { file := ` id__ID,s__String_F_YMDH,ts__Timestamp_s_2006-01-02 15:04:05.999,price__Decimal_2,age__Int_1_120 -0,a,1832-01-03 08:00:00.000,0.0,1 -1,b,1700-01-03 08:00:00.000,5.44,35 +0,a,0000-01-03 08:00:00.000,0.0,1 +1,b,9999-12-31 23:59:60.999,5.44,35 2,b,2019-50-03 08:00:00.000,5.44,120 3,b,2019-01-50 08:00:00.000,5.44,120 4,b,2019-01-03 50:00:00.000,5.44,120 @@ -328,7 +328,7 @@ id__ID,s__String_F_YMDH,ts__Timestamp_s_2006-01-02 15:04:05.999,price__Decimal_2 name := writeTempFile(t, file) checker := make(map[string]interface{}) - checker["ts"] = []interface{}{nil, nil, nil, nil, nil, nil, nil, "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", nil, "2019-04-03T00:00:00Z"} + checker["ts"] = []interface{}{nil, nil, nil, nil, nil, nil, nil, "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "2019-04-03T00:00:00Z", "1500-04-03T00:00:00Z", "2019-04-03T00:00:00Z"} checker["age"] = []interface{}{1, 35, 120, 120, 120, nil, 120, 1, 1, 100, nil, nil, nil, 100} checker["price"] = []interface{}{0.0, 5.44, 5.44, 5.44, 5.44, 5.44, 5.44, 123.12, -1.0, nil, 2.34, 3.44, nil, 3.44} @@ -415,6 +415,133 @@ id__ID,s__String_F_YMDH,ts__Timestamp_s_2006-01-02 15:04:05.999,price__Decimal_2 } +type TimestampTestCase struct { + fieldName string + file string + expect interface{} +} + +// Test Time Layouts +func TestTimeLayouts(t *testing.T) { + testCases := []TimestampTestCase{ + { + fieldName: "ts1", + file: ` +id__ID,s__String_F_YMDH,ts1__Timestamp_s_2006-01-02 15:04:05.999_2030-01-02 15:04:05.999_s +0,a,99221100 + +`[1:], + expect: []interface{}{"2033-02-24T00:29:05Z"}, + }, + { + fieldName: "ts-nano-min", + file: ` +id__ID,s__String_F_YMDH,ts-nano-min__Timestamp_ns_2006-01-02T15:04:05.999999999Z_1833-11-24T17:31:44.01Z_s +0,a,1 +1,b,-1 +`[1:], + expect: []interface{}{"1833-11-24T17:31:45.01Z", nil}, + }, + { + fieldName: "ts-nano-max", + file: ` +id__ID,s__String_F_YMDH,ts-nano-max__Timestamp_ns_2006-01-02T15:04:05.999999999Z_2106-02-07T06:28:16Z_ns +0,a,1 +1,b,-1000001 +`[1:], + expect: []interface{}{nil, "2106-02-07T06:28:15.998999999Z"}, + }, + { + fieldName: "ts-sec-min", + file: ` +id__ID,s__String_F_YMDH,ts-sec-min__Timestamp_s_2006-01-02T15:04:05.999999999Z_0001-01-01T00:00:01Z_ms +0,a,1001 +1,b,-1001 +`[1:], + expect: []interface{}{"0001-01-01T00:00:02Z", nil}, + }, + { + fieldName: "ts-ms-max", + file: ` +id__ID,s__String_F_YMDH,ts-ms-max__Timestamp_ms_2006-01-02T15:04:05.999999999Z_9999-12-31T23:59:59Z_us +0,a,1001 +1,b,-1001 + +`[1:], + expect: []interface{}{nil, "9999-12-31T23:59:58.999Z"}, + }, + { + fieldName: "gran-conversion", + file: ` +id__ID,s__String_F_YMDH,gran-conversion__Timestamp_ns_2006-01-02T15:04:05.999999999Z_2000-02-07T06:28:16Z_s +0,a,10000000000 +1,b,-1001 + +`[1:], + expect: []interface{}{nil, "2000-02-07T06:11:35Z"}, + }, + } + + for _, tc := range testCases { + m := newMainOORFactory(t, tc.file, false, false, true) + testTimestampRunner(t, m, tc) + } +} + +func testTimestampRunner(t *testing.T, m *Main, testCase TimestampTestCase) { + defer func() { + if err := m.PilosaClient().DeleteIndexByName(m.Index); err != nil { + t.Logf("deleting test index: %v", err) + } + }() + + err := m.Run() + if err != nil { + t.Fatalf("running: %v", err) + } + + pql := fmt.Sprintf("Extract(All(), Rows(%s))", testCase.fieldName) + + eResp, err := idktest.DoExtractQuery(pql, m.Index) + if err != nil { + t.Fatal("doing extract: ", err) + } + if eResp.Results[0].Columns == nil { + t.Fatal("no results: ", err) + } + + for i, item := range eResp.Results[0].Columns { + check := testCase.expect.([]interface{}) + switch exp := check[i].(type) { + case nil: + if item.Rows[0] != nil { + t.Errorf("expected nil, got %+v", item.Rows[0]) + } + case string: + if item.Rows[0] != exp { + t.Errorf("expected %s, got %+v", exp, item.Rows[0]) + } + case int: + vAsInt := int(item.Rows[0].(float64)) + if vAsInt != exp { + t.Errorf("expected %d, got %+v", exp, item.Rows[0]) + } else { + expAsFloat := float64(exp) + if item.Rows[0] != expAsFloat { + t.Errorf("expected %f, got %+v", expAsFloat, item.Rows[0]) + } + } + case float64: + if item.Rows[0] != exp { + t.Errorf("expected %f, got %+v", exp, item.Rows[0]) + } + default: + t.Errorf("unknown type: %T", exp) + } + } + +} + // Test that out of range int values are ingested as nil when AllowIntOutOfRange is true. func TestIntOpts(t *testing.T) { file := ` @@ -502,22 +629,22 @@ id__ID,negneg__Int_-10_-5,negpos__Int_-10_10,pospos__Int_5_10,negzero__Int_-10_0 // Test that out of range timestamp values are ingested as nil when AllowTimestampOutOfRange is true. func TestTimestampOOR(t *testing.T) { file := ` -id__ID,ts1__Timestamp_s_2006-01-02 15:04:05.999,ts2__Timestamp_s_2006-01-02T15:04:05Z07:00_2261-12-31T15:04:05Z_h,ts3__Timestamp_s_2006-01-02T15:04:05Z07:00_1679-12-31T15:04:05Z_h,ts4__Timestamp_s_2006-01-02 15:04:05.999 -0,1833-01-03 08:00:00.000,2431,-19960,2009-11-24 17:31:44.000 -1,1833-11-24 17:31:44.000,2433,-19955,1800-11-24 17:31:44.000 -2,1833-11-25 17:31:44.000,9999,0,2011-11-25 17:31:44.000 -3,2106-02-06 06:28:16.000,0,-99999,2008-02-06 06:28:16.000 -4,2106-02-07 06:28:16.000,9999,-99999,1800-02-07 06:28:16.000 -5,2106-02-08 06:28:16.000,99999999999999999999999,-9999999999999999999999999,2012-02-07 06:28:16.000 +id__ID,ts1__Timestamp_ns_2006-01-02 15:04:05.999,ts2__Timestamp_s_2006-01-02T15:04:05Z07:00_9998-12-31T15:04:05Z_h,ts3__Timestamp_s_2006-01-02T15:04:05Z07:00_0002-12-31T15:04:05Z_h,ts4__Timestamp_s_2006-01-02T15:04:05.999Z +0,1833-01-03 08:00:00.000,8500,8500,0001-01-01T00:00:00Z +1,1833-11-24 17:31:44.000,8769,-8500,0001-01-01T00:00:01Z +2,1833-11-25 17:31:44.000,-99991,0,0001-01-01T00:00:02Z +3,2106-02-06 06:28:16.000,0,-99995,9999-12-31T23:59:58Z +4,2106-02-07 06:28:16.000,9999,-99999,9999-12-31T23:59:59Z +5,2106-02-08 06:28:16.000,99999999999999999999999,-9999999999999999999999999,9999-12-31T23:59:60Z `[1:] checker := make(map[string]interface{}) // should import nil if timestamp val is out of range checker["ts1"] = []interface{}{nil, "1833-11-24T17:31:44Z", "1833-11-25T17:31:44Z", "2106-02-06T06:28:16Z", "2106-02-07T06:28:16Z", nil} // should import nil if custom epoch + value overflows - checker["ts2"] = []interface{}{"2262-04-11T22:04:05Z", nil, nil, "2261-12-31T15:04:05Z", nil, nil} - checker["ts3"] = []interface{}{nil, "1677-09-21T04:04:05Z", "1679-12-31T15:04:05Z", nil, nil, nil} - checker["ts4"] = []interface{}{"2009-11-24T17:31:44Z", nil, "2011-11-25T17:31:44Z", "2008-02-06T06:28:16Z", nil, "2012-02-07T06:28:16Z"} + checker["ts2"] = []interface{}{"9999-12-20T19:04:05Z", nil, "9987-08-05T08:04:05Z", "9998-12-31T15:04:05Z", nil, nil} + checker["ts3"] = []interface{}{"0003-12-20T19:04:05Z", "0002-01-11T11:04:05Z", "0002-12-31T15:04:05Z", nil, nil, nil} + checker["ts4"] = []interface{}{nil, "0001-01-01T00:00:01Z", "0001-01-01T00:00:02Z", "9999-12-31T23:59:58Z", "9999-12-31T23:59:59Z", nil} batchSizes := []int{3, 1, 4, 10} for _, bsize := range batchSizes { @@ -580,13 +707,13 @@ func TestFailureConditions(t *testing.T) { } testCases := []testCase{ - {name: "too small", csv: `id__ID,ts1__Timestamp_s_2006-01-02T15:04:05Z07:00_1600-12-31T15:04:05Z_h + {name: "too small", csv: `id__ID,ts1__Timestamp_s_2006-01-02T15:04:05Z07:00_0000-01-01T00:00:00Z_h 0,0 `, fail: true, intOutOfRange: true, timestampOutOfRange: true, decimalOutOfRange: true}, {name: "just right", csv: `id__ID,ts1__Timestamp_s_2006-01-02T15:04:05Z07:00_2200-12-31T15:04:05Z_h 0,0 `, fail: false, intOutOfRange: true, timestampOutOfRange: true, decimalOutOfRange: true}, - {name: "too big", csv: `id__ID,ts1__Timestamp_s_2006-01-02T15:04:05Z07:00_2262-12-31T15:04:05Z_h + {name: "too big", csv: `id__ID,ts1__Timestamp_s_2006-01-02T15:04:05Z07:00_9999-12-31T23:59:60Z_h 0,0 `, fail: true, intOutOfRange: true, timestampOutOfRange: true, decimalOutOfRange: true}, {name: "intOutOfRange not allowed", csv: `id__ID,pospos__Int_5_10 @@ -596,9 +723,9 @@ func TestFailureConditions(t *testing.T) { 0,11 `, fail: true, intOutOfRange: false, timestampOutOfRange: true, decimalOutOfRange: true}, {name: "timestampOutOfRange not allowed", csv: `id__ID,ts1__Timestamp_s_2006-01-02 15:04:05.999 -0,1833-01-03 08:00:00.000 +0,-0001-01-03 08:00:00.000 `, fail: true, intOutOfRange: true, timestampOutOfRange: false, decimalOutOfRange: true}, - {name: "timestampOutOfRange not allowed-2", csv: `id__ID,ts2__Timestamp_s_2006-01-02T15:04:05Z07:00_2261-12-31T15:04:05Z_h + {name: "timestampOutOfRange not allowed-2", csv: `id__ID,ts2__Timestamp_s_2006-01-02T15:04:05Z07:00_9999-12-31T23:59:59Z_h 0,2433 `, fail: true, intOutOfRange: true, timestampOutOfRange: false, decimalOutOfRange: true}, {name: "decimalOutOfRange not allowed", csv: `id__ID,price__Decimal_2 diff --git a/idk/header.go b/idk/header.go index ae13016e7..17abf65e4 100644 --- a/idk/header.go +++ b/idk/header.go @@ -275,7 +275,11 @@ func HeaderToField(headerField string, log logger.Logger) (field Field, _ error) if err != nil { return nil, errors.Wrapf(err, "parsing epoch for '%s'", headerField) } - tsField.Epoch = epoch + if epoch.IsZero() { + tsField.Epoch = time.Unix(0, 0) + } else { + tsField.Epoch = epoch + } } if len(fieldspec) > 4 { unit := Unit(fieldspec[4]).unit() diff --git a/idk/ingest.go b/idk/ingest.go index 1255ab092..cbd5d7c50 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -1430,6 +1430,7 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor } fields := make([]*pilosaclient.Field, 0, len(schema)) + existingFields := m.index.Fields() for i, idkField := range schema { // we redefine these inside the loop since we're // capturing them in closures @@ -1455,10 +1456,10 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor // different places _, isBool := idkField.(BoolField) _, isSIBK := idkField.(SignedIntBoolKeyField) - if !isSIBK && (m.PackBools == "" || !isBool) && m.index.HasField(idkField.DestName()) { + pilosaField, ok := existingFields[idkField.DestName()] + if !isSIBK && (m.PackBools == "" || !isBool) && ok { // Validate that Pilosa's existing field matches the // type and options of the IDK field. - pilosaField := m.index.Field(idkField.DestName()) if err := m.checkFieldCompatibility(pilosaField, idkField, ""); err != nil { return nil, nil, nil, nil, errors.Wrap(err, "checking field compatibility") } @@ -1638,7 +1639,7 @@ func (m *Main) batchFromSchema(schema []Field) ([]Recordizer, pilosaclient.Recor return errors.Wrapf(err, "converting field %d:%+v, val:%+v", i, idkField, rawRec[i]) }) case TimestampField: - fields = append(fields, m.index.Field(fld.DestName(), pilosaclient.OptFieldTypeTimestamp(fld.epoch(), fld.granularity()))) + fields = append(fields, m.index.Field(fld.DestName(), pilosaclient.OptFieldTypeTimestamp(fld.epoch(), string(fld.granularity())))) valIdx := len(fields) - 1 recordizers = append(recordizers, func(rawRec []interface{}, rec *pilosaclient.Row) (err error) { switch rawRec[i].(type) { diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 14ae7d43c..4d2f5ecd8 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -708,7 +708,7 @@ func TestBatchFromSchema(t *testing.T) { runTest := func(t *testing.T, test testcase, removeIndex bool, server serverInfo) { m := NewMain() configureTestFlags(m) - m.Index = "cmd_test_index23lkjdkfj" + m.Index = "cmd_test_index23lkjdkfjr2" m.PrimaryKeyFields = test.pkFields m.IDField = test.IDField m.AutoGenerate = test.autogen @@ -721,14 +721,6 @@ func TestBatchFromSchema(t *testing.T) { m.AuthToken = server.AuthToken } m.PilosaHosts = server.PilosaHosts - onFinishRun, err := m.Setup() - if strings.Contains("validation", test.name) && testErr(t, test.err, err) { - return - } - if err != nil { - t.Fatalf("%v", err) - } - defer onFinishRun() if removeIndex { defer func() { err := m.client.DeleteIndex(m.index) @@ -737,6 +729,14 @@ func TestBatchFromSchema(t *testing.T) { } }() } + onFinishRun, err := m.Setup() + if strings.Contains("validation", test.name) && testErr(t, test.err, err) { + return + } + if err != nil { + t.Fatalf("%v", err) + } + defer onFinishRun() rdzs, batch, row, lookupWriteIdxs, err := m.batchFromSchema(test.schema) if testErr(t, test.err, err) { @@ -937,7 +937,7 @@ func TestBatchFromSchema(t *testing.T) { rawRec: []interface{}{"blaah", "08 Mar 09 21:00 UTC", "molecula.com"}, rowID: "blaah", rowVals: []interface{}{"molecula.com"}, - time: getQuantizedTime(time.Unix(1236548950, 0)), + time: getQuantizedTime(time.Unix(1236548940, 0).UTC()), }, { name: "record time field epoch", @@ -959,7 +959,7 @@ func TestBatchFromSchema(t *testing.T) { }, { name: "timestamp field", - schema: []Field{StringField{NameVal: "a"}, TimestampField{NameVal: "b", Granularity: "s", Unit: Millisecond}}, + schema: []Field{StringField{NameVal: "a"}, TimestampField{NameVal: "b", Granularity: "s", Unit: Millisecond, Epoch: time.Unix(10000, 0)}}, pkFields: []string{"a"}, rawRec: []interface{}{"blah", "5000"}, rowID: "blah", @@ -967,7 +967,7 @@ func TestBatchFromSchema(t *testing.T) { }, { name: "timestamp incorrect layout", - schema: []Field{StringField{NameVal: "a"}, TimestampField{NameVal: "b", Granularity: "s", Layout: "Mon, 02 Jan 2006 15:04:05 MST"}}, + schema: []Field{StringField{NameVal: "a"}, TimestampField{NameVal: "b", Layout: "Mon, 02 Jan 2006 15:04:05 MST"}}, pkFields: []string{"a"}, rawRec: []interface{}{"blah", "2025-05-15T05:05:05Z"}, rowID: "blah", @@ -1173,18 +1173,17 @@ func int64Ptr(i int64) *int64 { func testErr(t *testing.T, exp string, actual error) (done bool) { t.Helper() - fmt.Printf("\n$$ Err Contains: %+v$$\n", actual) if exp == "" && actual == nil { return false } if exp == "" && actual != nil { - t.Fatalf("unexpected errs exp/got\n%s\n%v", exp, actual) + t.Fatalf("unexpected errs exp: \n%s got: \n%v", exp, actual) } if exp != "" && actual == nil { - t.Fatalf("expected errs exp/got\n%s\n%v", exp, actual) + t.Fatalf("expected errs exp: \n%s got: \n%v", exp, actual) } if !strings.Contains(actual.Error(), exp) { - t.Fatalf("unmatched errs exp/got\n%s\n%v", exp, actual) + t.Fatalf("unmatched errs exp: \n%s got: \n%v", exp, actual) } return true } diff --git a/idk/interfaces.go b/idk/interfaces.go index 6479ec794..935760da4 100644 --- a/idk/interfaces.go +++ b/idk/interfaces.go @@ -315,6 +315,13 @@ func (b BoolField) PilosafyVal(val interface{}) (interface{}, error) { return toBool(val) } +var ( + MinTimestampNano = time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z + MaxTimestampNano = time.Unix(1<<32, 0).UTC() // 2106-02-07T06:28:16Z + MinTimestamp = time.Unix(-62135596799, 0).UTC() // 0001-01-01T00:00:01Z + MaxTimestamp = time.Unix(253402300799, 0).UTC() // 9999-12-31T23:59:59Z +) + const ( Custom = Unit("c") Day = Unit("d") @@ -369,6 +376,34 @@ func (u Unit) Duration() (time.Duration, error) { return 0, errors.Errorf(ErrFmtUnknownUnit, u) } +// ToNanos returns the number of Nanoseconds per given Unit +func (u Unit) ToNanos() (int64, error) { + duration := int64(1) + switch u.unit() { + case Day: + duration *= 24 + fallthrough + case Hour: + duration *= 60 + fallthrough + case Minute: + duration *= 60 + fallthrough + case Second: + duration *= 1000 + fallthrough + case Millisecond: + duration *= 1000 + fallthrough + case Microsecond: + duration *= 1000 + fallthrough + case Nanosecond: + return duration, nil + } + return 0, errors.Errorf(ErrFmtUnknownUnit, u) +} + func (u Unit) DurationFromValue(val int64) (time.Duration, error) { scale, err := u.Duration() if err != nil { @@ -426,6 +461,9 @@ func (r RecordTimeField) PilosafyVal(val interface{}) (interface{}, error) { if err != nil { err = errors.Wrap(err, "converting RecordTimeField from layout") } + if result.IsZero() { + return nil, err + } return result, err } @@ -742,25 +780,68 @@ func (d TimestampField) DestName() string { return d.DestNameVal } +// 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 string(Second): + return time.Unix(val, 0).UTC(), nil + case string(Millisecond): + return time.UnixMilli(val).UTC(), nil + case string(Microsecond): + return time.UnixMicro(val).UTC(), nil + case string(Nanosecond): + return time.Unix(0, val).UTC(), nil + default: + return time.Time{}, errors.Errorf("Unknown time unit: '%v'", unit) + } +} + +// TimestampToVal takes a time unit and a time.Time and converts it to an integer value +func TimestampToVal(unit Unit, ts time.Time) int64 { + switch unit { + case Second: + return ts.Unix() + case Millisecond: + return ts.UnixMilli() + case Microsecond: + return ts.UnixMicro() + case Nanosecond: + return ts.UnixNano() + } + return 0 + +} + // PilosafyVal for TimestampField always returns an int or nil. func (t TimestampField) PilosafyVal(val interface{}) (interface{}, error) { - var dur time.Duration - if val == nil { return nil, nil } - granAsDur, err := Unit(t.granularity()).DurationFromValue(1) //return ns per granularity - if err != nil { - return nil, errors.Wrap(err, "converting granularity to duration") + var dur int64 + // Check if the epoch alone is out-of-range. If so, ingest should halt, regardless + // of state of the timestamp out-of-range CLI option. + if err := validateTimestamp(t.granularity(), t.epoch()); err != nil { + return nil, errors.Wrap(err, "validating epoch") } - if tval, ok := val.(time.Time); ok { - if err := t.validateTimestamp(tval); err != nil { - return nil, errors.Wrap(err, "validating timestamp") + epochAsVal := TimestampToVal(t.granularity(), t.epoch()) + if _, ok := val.(time.Time); ok || (t.Epoch.IsZero() && t.Unit == "") { + ts, err := timeFromTimestring(val, t.layout()) + if err != nil { + if strings.Contains(err.Error(), "out of range") { + return nil, errors.Wrap(err, ErrTimestampOutOfRange.Error()) + } + return nil, errors.Wrap(err, "converting TimestampField from layout") } - dur = tval.Sub(t.epoch()) - } else if !t.Epoch.IsZero() || t.Unit != "" { + if err := validateTimestamp(t.granularity(), ts); err != nil { + return nil, errors.Wrap(ErrTimestampOutOfRange, "validating timestamp") + } + + tsAsVal := TimestampToVal(t.granularity(), ts) + + dur = tsAsVal - epochAsVal + } else { valAsInt, err := toInt64(val) if err != nil { if strings.Contains(err.Error(), "out of range") { @@ -768,44 +849,46 @@ func (t TimestampField) PilosafyVal(val interface{}) (interface{}, error) { } return nil, errors.Wrap(err, "converting value to int64") } - dur, err = t.Unit.DurationFromValue(valAsInt) - if err != nil { - return nil, errors.Wrap(err, "converting TimestampField from epoch") + + // Conversion ratio to scale incoming Units to Granularity + granNanos, err := Unit(t.granularity()).ToNanos() + if err != nil || granNanos == 0 { + return nil, errors.Wrap(err, "granularity not supported") } - if err := t.validateDuration(dur, granAsDur); err != nil { + unitNanos, err := Unit(t.Unit).ToNanos() + if err != nil { + return nil, errors.Wrap(err, "unit not supported") + } + scale := float64(unitNanos) / float64(granNanos) + + dur = int64(float64(valAsInt) * scale) + if (dur >= 0 && valAsInt < 0) || (dur < 0 && valAsInt > 0) { + return nil, errors.Wrap(ErrTimestampOutOfRange, "timestamp value out of range at specified granularity") + } + + if err := validateDuration(dur, epochAsVal, Unit(t.granularity())); err != nil { return nil, errors.Wrap(err, "validating duration") } - } else { - ti, err := timeFromTimestring(val, t.layout()) - if err != nil { - if strings.Contains(err.Error(), "out of range") { - return nil, errors.Wrap(err, ErrTimestampOutOfRange.Error()) - } - return nil, errors.Wrap(err, "converting TimestampField from layout") - } - // if timeFromTimestring is nil, the time is either nil or "" - // and there is no time value to write. So we return a nil. - if ti == nil { - return nil, nil - } - ts := ti.(time.Time) - if err := t.validateTimestamp(ts); err != nil { - return nil, errors.Wrap(err, "validating timestamp") - } - dur = ts.Sub(t.epoch()) } - return int64(dur / granAsDur), nil + return dur, nil } // validateTimestamp checks if the timestamp is within the range of what FB accepts. -func (t TimestampField) validateTimestamp(ts time.Time) error { +func validateTimestamp(unit Unit, ts time.Time) error { // Min and Max timestamps that Featurebase accepts - MinTimestamp := time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z - MaxTimestamp := time.Unix(1<<32, 0).UTC() // 2106-02-07T06:28:16Z + var minStamp, maxStamp time.Time + switch unit { + case Nanosecond: + minStamp = MinTimestampNano + maxStamp = MaxTimestampNano + default: + minStamp = MinTimestamp + maxStamp = MaxTimestamp + } - if ts.Before(MinTimestamp) || ts.After(MaxTimestamp) { - return errors.Wrap(ErrTimestampOutOfRange, fmt.Sprintf("timestamp value must be within min: %v and max: %v", MinTimestamp, MaxTimestamp)) + if ts.Before(minStamp) || ts.After(maxStamp) { + return errors.New(fmt.Sprintf("timestamp value must be within min: %v and max: %v", minStamp, maxStamp)) } return nil } @@ -814,34 +897,39 @@ func (t TimestampField) validateTimestamp(ts time.Time) error { // Featurebase will ultimately convert this to some duration relative to the Unix epoch. // So if the custom epoch + the provided value in the desired units is too far from // Unix epoch such that it causes an interger overflow, this will return an error. -func (t TimestampField) validateDuration(dur time.Duration, granularity time.Duration) error { - // Check if the epoch alone causes overflow. If so, ingest should halt. - if time.Unix(0, 0).After(t.epoch()) { - if t.epoch().UnixNano() > 0 { - return errors.New("custom epoch is too far from Unix epoch") - } - } else if t.epoch().UnixNano() < 0 { - return errors.New("custom epoch is too far from Unix epoch") +func validateDuration(dur int64, offset int64, granularity Unit) error { + var minInt, maxInt int64 + switch granularity { + case Second: + minInt = MinTimestamp.Unix() + maxInt = MaxTimestamp.Unix() + case Millisecond: + minInt = MinTimestamp.UnixMilli() + maxInt = MaxTimestamp.UnixMilli() + case Microsecond: + minInt = MinTimestamp.UnixMicro() + maxInt = MaxTimestamp.UnixMicro() + case Nanosecond: + minInt = MinTimestampNano.UnixNano() + maxInt = MaxTimestampNano.UnixNano() } - offset := int64(t.epoch().Sub(time.Unix(0, 0)).Nanoseconds()) - durAsInt := int64(dur) if offset > 0 { - if durAsInt > math.MaxInt64-offset { + if dur > maxInt-offset { return errors.Wrap(ErrTimestampOutOfRange, "value + epoch is too far from Unix epoch") } - } else if durAsInt < math.MinInt64-offset { + } else if dur < minInt-offset { return errors.Wrap(ErrTimestampOutOfRange, "value + epoch is too far from Unix epoch") } return nil } // Return default granularity if not set -func (t TimestampField) granularity() string { +func (t TimestampField) granularity() Unit { if t.Granularity == "" { return "s" } - return t.Granularity + return Unit(t.Granularity) } // Return default layout if not set @@ -966,35 +1054,35 @@ func timeFromEpoch(val interface{}, epoch time.Time, unit Unit) (interface{}, er return epoch.Add(dur), err } -func timeFromTimestring(val interface{}, layout string) (interface{}, error) { +func timeFromTimestring(val interface{}, layout string) (time.Time, error) { if val == nil { - return nil, nil + return time.Time{}, nil } switch valt := val.(type) { case nil: - return nil, nil + return time.Time{}, nil case []byte: if len(valt) == 0 { - return nil, nil + return time.Time{}, nil } vt, err := parseTimeWithLayout(layout, string(valt)) if err != nil { - return nil, errors.Wrap(err, "parsing []byte") + return time.Time{}, errors.Wrap(err, "parsing []byte") } return vt, nil case string: if valt == "" { - return nil, nil + return time.Time{}, nil } vt, err := parseTimeWithLayout(layout, valt) if err != nil { - return nil, errors.Wrapf(err, "parsing time string %s", valt) + return time.Time{}, errors.Wrapf(err, "parsing time string %s", valt) } return vt, nil case time.Time: return valt, nil default: - return nil, errors.Errorf("didn't know how to interpret %v of %[1]T as time", valt) + return time.Time{}, errors.Errorf("didn't know how to interpret %v of %[1]T as time", valt) } } diff --git a/idk/interfaces_test.go b/idk/interfaces_test.go index e6afb53cb..c5b4a2002 100644 --- a/idk/interfaces_test.go +++ b/idk/interfaces_test.go @@ -239,3 +239,198 @@ func TestTTLOf(t *testing.T) { } } + +func TestTimestampField_validateDuration(t *testing.T) { + type args struct { + dur int64 + offset int64 + unit Unit + } + tests := []struct { + name string + args args + wantErr bool + errStr string + }{ + { + name: "max-max", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("s"), MaxTimestamp), + unit: Unit("s"), + }, + wantErr: false, + }, + { + name: "oor-max", + args: args{ + dur: 1, + offset: TimestampToVal(Unit("s"), MaxTimestamp), + unit: Unit("s"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + { + name: "max-oor", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("s"), MaxTimestamp.Add(1*time.Second)), + unit: Unit("s"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + { + name: "min-min", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("s"), MinTimestamp), + unit: Unit("s"), + }, + wantErr: false, + }, + { + name: "oor-min", + args: args{ + dur: -1, + offset: TimestampToVal(Unit("s"), MinTimestamp), + unit: Unit("s"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + { + name: "min-oor", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("s"), MinTimestamp.Add(-1*time.Second)), + unit: Unit("s"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + { + name: "max-max", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("ns"), MaxTimestampNano), + unit: Unit("ns"), + }, + wantErr: false, + }, + { + name: "oor-max", + args: args{ + dur: 1, + offset: TimestampToVal(Unit("ns"), MaxTimestampNano), + unit: Unit("ns"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + { + name: "max-oor", + args: args{ + dur: 0, + offset: TimestampToVal(Unit("ns"), MaxTimestampNano.Add(1*time.Nanosecond)), + unit: Unit("ns"), + }, + wantErr: true, + errStr: "value + epoch is too far from Unix epoch", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := validateDuration(tt.args.dur, tt.args.offset, tt.args.unit); (err != nil) != tt.wantErr { + if !strings.Contains(err.Error(), tt.errStr) { + t.Errorf("TimestampField.validateDuration() error = %v, wantErr %v", err, tt.wantErr) + } + } + }) + } +} + +func Test_validateTimestamp(t *testing.T) { + type args struct { + unit Unit + ts time.Time + } + tests := []struct { + name string + args args + wantErr bool + }{ + { + name: "nano-max", + args: args{ + unit: Unit("ns"), + ts: MaxTimestampNano, + }, + wantErr: false, + }, + { + name: "nano-min", + args: args{ + unit: Unit("ns"), + ts: MinTimestampNano, + }, + wantErr: false, + }, + { + name: "nano-max-oor", + args: args{ + unit: Unit("ns"), + ts: MaxTimestampNano.Add(1 * time.Nanosecond), + }, + wantErr: true, + }, + { + name: "nano-min-oor", + args: args{ + unit: Unit("ns"), + ts: MinTimestampNano.Add(-1 * time.Nanosecond), + }, + wantErr: true, + }, + { + name: "ms-max", + args: args{ + unit: Unit("ms"), + ts: MaxTimestamp, + }, + wantErr: false, + }, + { + name: "ms-min", + args: args{ + unit: Unit("ms"), + ts: MinTimestamp, + }, + wantErr: false, + }, + { + name: "ms-max-oor", + args: args{ + unit: Unit("ms"), + ts: MaxTimestamp.Add(1 * time.Millisecond), + }, + wantErr: true, + }, + { + name: "ms-min-oor", + args: args{ + unit: Unit("ms"), + ts: MinTimestamp.Add(-1 * time.Millisecond), + }, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := validateTimestamp(tt.args.unit, tt.args.ts); (err != nil) != tt.wantErr { + t.Errorf("validateTimestamp() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +} 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<