From 3681feeeb2d242c4b3fd63c168da9c4e5b7d2471 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Thu, 4 Aug 2022 17:20:45 -0500 Subject: [PATCH] [FB-1024 FB-1590] Increase timerange (un-revert) (#2174) previously we allowed users to specify a granularity for timestamp e.g. seconds, milli, micro, nano however we converted everything to nano before we stored it. This reduced the allowed range for all time units to what was allowed by timestamp. For example, with second granularity you can represent billions of years within the capacity of int64 but with nano its somewhere b/w 100-200 years. So now, for timeunits of seconds, milli, and micro the range is year 0001 - 9999. These limits come from what Go supports. So this uses unit specific function to translate timestamps to values and vice versa to increase the time range. In the process of increasing the range for timestamp and subsequent testing, I found and addressed a few bugs: - min/max queries were not using timestamp specific comparators so added that. - Values from Import/ingest come to FB as relative values to epoch whereas other BSI fields come as actual values and then becomes relative to their respective bases within FB. so some specific handling of that was added. - However! Set queries use timestamp strings which are, of course, the actual value they designate. So they have to become relative. - When bitdepth is 0, Min/maxUnsigned functions did not run resulting in a count of 0 when there was an actual value that was 0. Also, this removes (now) dead code and updates/adds tests. --- .gitlab/.gitlab-ci.yml | 4 +- api.go | 3 +- client/client.go | 2 + executor.go | 118 +++++++- executor_test.go | 625 ++++++++++++++++++++++++++++++++++++++++- field.go | 147 ++++++---- field_test.go | 12 +- fragment.go | 2 + idk/csv/csv_test.go | 161 +++++++++-- idk/header.go | 6 +- idk/ingest.go | 7 +- idk/ingest_test.go | 31 +- idk/interfaces.go | 208 ++++++++++---- idk/interfaces_test.go | 195 +++++++++++++ ingest/codec.go | 54 +++- ingest/codec_test.go | 8 +- server/pg.go | 7 +- util.go | 6 - util_test.go | 18 -- 19 files changed, 1395 insertions(+), 219 deletions(-) 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<