diff --git a/api.go b/api.go index 400deb995..ca4db1c9a 100644 --- a/api.go +++ b/api.go @@ -2006,7 +2006,10 @@ 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) // TODO: Samir tagging this for futher inspection + // TODO: codec will likely need to be updated to reflect changes (increase) to timestamp range + // i.e. values are not all being converted to Nanos and being stored at the granularity + // specified by user and taking advantage of the range that provides. + nanos := TimeUnitNanos(field.options.TimeUnit) if err = codec.AddTimestampField(field.name, time.Duration(nanos), field.options.Base); err != nil { return fmt.Errorf("adding timestamp field to codec: %w", err) } diff --git a/api_test.go b/api_test.go index 0d5fc72f3..13ec8f8ae 100644 --- a/api_test.go +++ b/api_test.go @@ -422,55 +422,53 @@ func TestAPI_ImportValue(t *testing.T) { } }) - // Needs to be updated with new min max timestamp values + t.Run("ValTimestampField", func(t *testing.T) { + t.Skip() // skipping due to change partitioning strategy + ctx := context.Background() + index := "valts" + field := "fts" - // t.Run("ValTimestampField", func(t *testing.T) { - // t.Skip() // skipping due to change partitioning strategy - // ctx := context.Background() - // index := "valts" - // field := "fts" + _, err := m1.API.CreateIndex(ctx, index, pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index: %v", err) + } + _, err = m1.API.CreateField(ctx, index, field, pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)) + if err != nil { + t.Fatalf("creating field: %v", err) + } - // _, err := m1.API.CreateIndex(ctx, index, pilosa.IndexOptions{}) - // if err != nil { - // t.Fatalf("creating index: %v", err) - // } - // _, err = m1.API.CreateField(ctx, index, field, pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)) - // if err != nil { - // t.Fatalf("creating field: %v", err) - // } + // Generate some records. + values := []time.Time{} + colIDs := []uint64{} + for i := 0; i < 10; i++ { + values = append(values, pilosa.MinTimestamp.Add(time.Duration(i)*time.Second)) + colIDs = append(colIDs, uint64(i)) + } - // // Generate some records. - // values := []time.Time{} - // colIDs := []uint64{} - // for i := 0; i < 10; i++ { - // values = append(values, pilosa.MinTimestamp.Add(time.Duration(i)*time.Second)) - // colIDs = append(colIDs, uint64(i)) - // } + // Import data with keys to node1 and verify that it gets translated and + // forwarded to the owner of shard 0 (node0; because of offsetModHasher) + req := &pilosa.ImportValueRequest{ + Index: index, + Field: field, + ColumnIDs: colIDs, + TimestampValues: values, + } - // // Import data with keys to node1 and verify that it gets translated and - // // forwarded to the owner of shard 0 (node0; because of offsetModHasher) - // req := &pilosa.ImportValueRequest{ - // Index: index, - // Field: field, - // ColumnIDs: colIDs, - // TimestampValues: values, - // } + qcx := m2.API.Txf().NewQcx() + if err := m2.API.ImportValue(ctx, qcx, req); err != nil { + t.Fatal(err) + } + PanicOn(qcx.Finish()) - // qcx := m2.API.Txf().NewQcx() - // if err := m2.API.ImportValue(ctx, qcx, req); err != nil { - // t.Fatal(err) - // } - // PanicOn(qcx.Finish()) + query := fmt.Sprintf("Row(%s>='1833-11-24T17:31:50Z')", field) // 6s after MinTimestamp - // query := fmt.Sprintf("Row(%s>='1833-11-24T17:31:50Z')", field) // 6s after MinTimestamp - - // // Query node0. - // if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: query}); err != nil { - // t.Fatal(err) - // } else if ids := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(ids, colIDs[6:]) { - // t.Fatalf("unexpected column keys: observerd %+v; expected '%+v'", ids, colIDs[6:]) - // } - // }) + // Query node0. + if res, err := m0.API.Query(ctx, &pilosa.QueryRequest{Index: index, Query: query}); err != nil { + t.Fatal(err) + } else if ids := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(ids, colIDs[6:]) { + t.Fatalf("unexpected column keys: observerd %+v; expected '%+v'", ids, colIDs[6:]) + } + }) t.Run("ValStringField", func(t *testing.T) { t.Skip() // skipping due to change partitioning strategy diff --git a/executor.go b/executor.go index a8e4fbb35..d3eb8a07a 100644 --- a/executor.go +++ b/executor.go @@ -1632,6 +1632,7 @@ 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)) } @@ -3205,7 +3206,6 @@ func (fr FieldRow) MarshalJSON() ([]byte, error) { if err != nil { return nil, errors.Wrap(err, "translating value to timestamp") } - // ts := FormatTimestampNano(int64(*fr.Value), fr.FieldOptions.Base, fr.FieldOptions.TimeUnit) return json.Marshal(struct { Field string `json:"field"` Value string `json:"value"` @@ -7580,6 +7580,7 @@ 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: @@ -7595,6 +7596,7 @@ func ValToTimestamp(unit string, val int64) (time.Time, error) { } } +// 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: @@ -7917,6 +7919,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.Equal(time.Time{}) || !other.TimestampVal.Equal(time.Time{}) { + return vc.timestampSmaller(other) } if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) { return other @@ -7934,6 +7938,24 @@ func (vc *ValCount) smaller(other ValCount) ValCount { } } +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, + } +} + func (vc *ValCount) decimalSmaller(other ValCount) ValCount { if other.DecimalVal == nil { return *vc @@ -7971,6 +7993,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 @@ -7988,6 +8012,24 @@ func (vc *ValCount) larger(other ValCount) ValCount { } } +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, + } +} + func (vc *ValCount) decimalLarger(other ValCount) ValCount { if other.DecimalVal == nil { return *vc @@ -8425,7 +8467,6 @@ func getScaledInt(f *Field, v interface{}) (int64, error) { case time.Time: v := TimestampToVal(f.options.TimeUnit, tv) value = v - // value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit) case int64: value = tv default: diff --git a/executor_test.go b/executor_test.go index d4b6821b7..f22c79c09 100644 --- a/executor_test.go +++ b/executor_test.go @@ -7226,24 +7226,24 @@ func TestMissingKeyRegression(t *testing.T) { // (single and multi-node clusters, different endpoints for the // queries (HTTP, GRPC, Postgres), etc.). func TestVariousQueries(t *testing.T) { - for _, clusterSize := range []int{3} { + for _, clusterSize := range []int{1, 3, 5} { clusterSize := clusterSize t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) { c := test.MustRunCluster(t, clusterSize) defer c.Close() // put a variety of data into the cluster - // populateTestData(t, c) - // backupTest(t, c, usersIndex) + populateTestData(t, c) + backupTest(t, c, usersIndex) - // variousQueries(t, c) - // variousQueriesOnTimeFields(t, c) - // variousQueriesOnPercentiles(t, c) - // variousQueriesCountDistinctTimestamp(t, c) - // variousQueriesOnIntFields(t, c) + variousQueries(t, c) + variousQueriesOnTimeFields(t, c) + variousQueriesOnPercentiles(t, c) + variousQueriesCountDistinctTimestamp(t, c) + variousQueriesOnIntFields(t, c) variousQueriesOnTimestampFields(t, c) variousQueriesOnLargeEpoch(t, c) - // backupTest(t, c, "") // test backup/restore of all indexes + backupTest(t, c, "") // test backup/restore of all indexes }) } } @@ -7493,6 +7493,10 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { } } +func toCSV(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 { @@ -7539,10 +7543,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) @@ -7703,49 +7703,46 @@ userG,-1,10,10,10 } } +// Constants used in TimestampField testing var ( minTime = pilosa.MinTimestamp maxTime = pilosa.MaxTimestamp - minSec = pilosa.MinTimestamp.Unix() - maxSec = pilosa.MaxTimestamp.Unix() - minMilli = pilosa.MinTimestamp.UnixMilli() - maxMilli = pilosa.MaxTimestamp.UnixMilli() - minMicro = pilosa.MinTimestamp.UnixMicro() - maxMicro = pilosa.MaxTimestamp.UnixMicro() + 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() ) -// tests queries on Timestamp Fields at various granularities +// variousQueriesOnTimestampFields tests queries on Timestamp Fields at various granularities using the default epoch func variousQueriesOnTimestampFields(t *testing.T, c *test.Cluster) { - index := "ts_test98" + index := "ts_test01" - // Testing whether the max and min values for seconds. Note this is less than min and max signed int 64 values - // since go has to account for the difference between Unix Epoch and Go's Epoch + // 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"}, - // {Val: 0, Key: "userC"}, - // {Val: 5000, Key: "userD"}, - // {Val: -5000, Key: "userE"}, }) - // Testing whether the max and min signed int 64 can be represented for milliseconds + // 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 signed int 64 can be represented for microseconds + // 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"}, }) - // Testing whether the max and min signed int 64 can be represented for nanoseconds + // 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"}, @@ -7761,10 +7758,6 @@ func variousQueriesOnTimestampFields(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) @@ -7774,19 +7767,19 @@ func variousQueriesOnTimestampFields(t *testing.T, c *test.Cluster) { tests := []testCase{ { query: "extract(All(), Rows(unix_sec))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,9999-12-31T23:59:59Z `, }, { query: "extract(All(), Rows(unix_milli))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,9999-12-31T23:59:59Z `, }, { query: "extract(All(), Rows(unix_micro))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,9999-12-31T23:59:59Z `, }, @@ -7817,13 +7810,13 @@ userB // }, // { // query: "Distinct(field=unix_milli)", - // csvVerifier: `0000-01-01T00:00:00Z + // csvVerifier: `0001-01-01T00:00:01Z // 9999-12-31T23:59:59Z // `, // }, { query: "Min(unix_sec)", - csvVerifier: `0000-01-01T00:00:00Z,1 + csvVerifier: `0001-01-01T00:00:01Z,1 `, }, { @@ -7833,7 +7826,7 @@ userB }, { query: "Min(unix_milli)", - csvVerifier: `0000-01-01T00:00:00Z,1 + csvVerifier: `0001-01-01T00:00:01Z,1 `, }, { @@ -7843,7 +7836,7 @@ userB }, { query: "Min(unix_micro)", - csvVerifier: `0000-01-01T00:00:00Z,1 + csvVerifier: `0001-01-01T00:00:01Z,1 `, }, { @@ -7863,12 +7856,12 @@ userB }, { query: "GroupBy(Rows(unix_micro))", - csvVerifier: `-62167219200000000,1 + csvVerifier: `-62135596799000000,1 253402300799000000,1 `, }, { - query: `Row(unix_sec="0000-01-01T00:00:00Z")`, + query: `Row(unix_sec="0001-01-01T00:00:01Z")`, csvVerifier: toCSV("userA"), }, { @@ -7876,7 +7869,7 @@ userB csvVerifier: toCSV("userB"), }, { - query: `Row(unix_milli="0000-01-01T00:00:00Z")`, + query: `Row(unix_milli="0001-01-01T00:00:01Z")`, csvVerifier: toCSV("userA"), }, { @@ -7884,7 +7877,7 @@ userB csvVerifier: toCSV("userB"), }, { - query: `Row(unix_micro="0000-01-01T00:00:00Z")`, + query: `Row(unix_micro="0001-01-01T00:00:01Z")`, csvVerifier: toCSV("userA"), }, { @@ -7900,7 +7893,7 @@ userB csvVerifier: toCSV("userB"), }, { - query: `Union(Row(unix_nano="2106-02-07T06:28:16Z"), Row(unix_micro="0000-01-01T00:00:00Z"))`, + query: `Union(Row(unix_nano="2106-02-07T06:28:16Z"), Row(unix_micro="0001-01-01T00:00:01Z"))`, csvVerifier: toCSV("userA\nuserB"), }, { @@ -7913,7 +7906,7 @@ userB csvVerifier: `true `, }, - // Not Yet Supported for Timestamp + // Not Supported for Timestamp // { // query: `ClearRow(unix_milli="2000-12-31T23:59:59.999Z")`, // csvVerifier: `true`, @@ -7922,11 +7915,7 @@ userB for i, tst := range tests { t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { - // resp := c.Query(t, index, tst.query) tr := c.QueryGRPC(t, index, tst.query) - // if tst.qrVerifier != nil { - // tst.qrVerifier(t, resp) - // } csvString, err := tableResponseToCSVString(tr) if err != nil { t.Fatal(err) @@ -7940,12 +7929,12 @@ userB } } -// tests queries on Timestamp Fields with large epochs +// 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_test2232" + index := "ts_epoch_test20321" + // mag := int64(10000000000) - posTime := time.Unix(0, 662687999999999999) - negTime := time.Unix(0, -662687999999999999) // 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{ @@ -7954,11 +7943,16 @@ func variousQueriesOnLargeEpoch(t *testing.T, c *test.Cluster) { {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"}, - {Val: -maxSec, Key: "userB"}, - {Val: -maxSec + minSec, Key: "userC"}, + {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")) @@ -7989,31 +7983,24 @@ func variousQueriesOnLargeEpoch(t *testing.T, c *test.Cluster) { {Val: -maxMicro + minMicro, Key: "userC"}, }) - c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_nano_neg", pilosa.OptFieldTypeTimestamp(negTime, "ns")) - c.ImportIntKey(t, index, "unix_nano_neg", []test.IntKey{ - {Val: 0, Key: "userA"}, - {Val: 662687999999999999, Key: "userB"}, - }) - - c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "unix_nano_pos", pilosa.OptFieldTypeTimestamp(posTime, "ns")) - c.ImportIntKey(t, index, "unix_nano_pos", []test.IntKey{ - {Val: 0, Key: "userA"}, - {Val: -662687999999999999, Key: "userB"}, - }) - 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: maxNano, Key: "userB"}, + {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: minNano, Key: "userB"}, + {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" @@ -8028,7 +8015,7 @@ func variousQueriesOnLargeEpoch(t *testing.T, c *test.Cluster) { tests := []testCase{ { query: "extract(All(), Rows(unix_sec_min))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,1970-01-01T00:00:00Z userC,9999-12-31T23:59:59Z `, @@ -8037,12 +8024,12 @@ 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,0000-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z `, }, { query: "extract(All(), Rows(unix_milli_min))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,1970-01-01T00:00:00Z userC,9999-12-31T23:59:59Z `, @@ -8051,12 +8038,12 @@ 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,0000-01-01T00:00:00Z +userC,0001-01-01T00:00:01Z `, }, { query: "extract(All(), Rows(unix_micro_min))", - csvVerifier: `userA,0000-01-01T00:00:00Z + csvVerifier: `userA,0001-01-01T00:00:01Z userB,1970-01-01T00:00:00Z userC,9999-12-31T23:59:59Z `, @@ -8065,46 +8052,236 @@ 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,0000-01-01T00:00:00Z -`, - }, - { - query: "extract(All(), Rows(unix_nano_neg))", - csvVerifier: `userA,1949-01-01T00:00:00.000000001Z -userB,1970-01-01T00:00:00Z -userC, -`, - }, - { - query: "extract(All(), Rows(unix_nano_pos))", - csvVerifier: `userA,1990-12-31T23:59:59.999999999Z -userB,1970-01-01T00:00:00Z -userC, +userC,0001-01-01T00:00:01Z `, }, { query: "extract(All(), Rows(unix_nano_min))", csvVerifier: `userA,1833-11-24T17:31:44Z -userB,1970-01-01T00:00:00Z -userC, +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: toCSV("userA"), + }, + { + query: `Row(unix_sec_max="0001-01-01T00:00:01Z")`, + csvVerifier: toCSV("userC"), + }, + { + query: `Row(unix_sec_max="9999-12-31T23:59:59Z")`, + csvVerifier: toCSV("userA"), + }, + { + query: `Row(unix_milli_min="0001-01-01T00:00:01Z")`, + csvVerifier: toCSV("userA"), + }, + { + query: `Row(unix_milli_min="9999-12-31T23:59:59Z")`, + csvVerifier: toCSV("userC"), + }, + { + query: `Row(unix_micro_max="0001-01-01T00:00:01Z")`, + csvVerifier: toCSV("userC"), + }, + { + query: `Row(unix_micro_max="9999-12-31T23:59:59Z")`, + csvVerifier: toCSV("userA"), + }, + { + query: `Row(unix_nano_min="1833-11-24T17:31:44Z")`, + csvVerifier: toCSV("userA"), + }, + { + query: `Row(unix_nano_min="1969-12-31T23:59:59.999999999Z")`, + csvVerifier: toCSV("userB"), + }, + { + query: `Row(unix_nano_min="1970-01-01T00:00:00Z")`, + csvVerifier: toCSV("userC"), + }, + { + query: `Row(unix_nano_max="2106-02-07T06:28:16Z")`, + csvVerifier: toCSV("userA"), + }, + { + query: `Row(unix_nano_max="1970-01-01T00:00:00.000000001Z")`, + csvVerifier: toCSV("userB"), + }, + { + query: `Row(unix_nano_max="1970-01-01T00:00:00Z")`, + csvVerifier: toCSV("userC"), + }, + { + query: `Union(Row(unix_nano_max="2106-02-07T06:28:16Z"), Row(unix_micro_max="0001-01-01T00:00:01Z"))`, + csvVerifier: toCSV("userA\nuserC"), + }, + { + 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, +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) { - resp := c.Query(t, index, tst.query) tr := c.QueryGRPC(t, index, tst.query) - if tst.qrVerifier != nil { - tst.qrVerifier(t, resp) - } csvString, err := tableResponseToCSVString(tr) if err != nil { t.Fatal(err) diff --git a/field.go b/field.go index 6276fbb6e..4e0606fef 100644 --- a/field.go +++ b/field.go @@ -217,6 +217,10 @@ func OptFieldTypeTimestamp(epoch time.Time, timeUnit string) FieldOption { 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 @@ -231,7 +235,7 @@ func OptFieldTypeTimestamp(epoch time.Time, timeUnit string) FieldOption { return errors.Errorf("invalid time unit: '%q'", fo.TimeUnit) } - if err := checkEpochOutOfRange(epoch, minTime, maxTime); err != nil { + if err := CheckEpochOutOfRange(epoch, minTime, maxTime); err != nil { return err } @@ -1431,14 +1435,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) @@ -1817,25 +1826,23 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard if value < min { min = value } - // if f.Type() == FieldTypeTimestamp { - // base := f.Options().Base - // if base > 0 { - // if value > f.Options().Max.ToInt64(0)-base { - // vprint.VV("value: %+v, Max: %+v, Base: %+v", value, f.Options().Max.ToInt64(0), base) - // return errors.Wrap(ErrBSIGroupValueTooHigh, "value + epoch is too far from Unix epoch") - // } - // } else if value < f.Options().Min.ToInt64(0)-base { - // vprint.VV("value: %+v, Min: %+v, Base: %+v", value, f.Options().Min.ToInt64(0), base) - // 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 coure 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. @@ -2157,16 +2164,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))) // TODO: Samir should be dead code now -// } - -// 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))) // TODO: Samir should be dead code now -// } - // List of bsiGroup types. const ( bsiGroupTypeInt = "int" @@ -2332,19 +2329,13 @@ 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 - // MinTimestampNano = time.Unix(0, MinNano) - // MaxTimestampNano = time.Unix(0, MaxNano) - + 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(-62167219200, 0).UTC() // 0000-01-01 00:00:00 +0000 UTC - MaxTimestamp = time.Unix(253402300799, 0).UTC() // 9999-12-31 23:59:59 +0000 UTC - - // MinTimestamp = time.Unix(-1<<32, 0).UTC() // 1833-11-24T17:31:44Z - // MaxTimestamp = 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 ) // Constants related to timestamp. @@ -2380,8 +2371,8 @@ func TimeUnitNanos(unit string) int64 { } } -// checkEpochOutOfRange checks if the epoch is after max or before min -func checkEpochOutOfRange(epoch, min, max time.Time) error { +// 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) } diff --git a/field_test.go b/field_test.go index 5043c6504..85299d4ea 100644 --- a/field_test.go +++ b/field_test.go @@ -6,6 +6,7 @@ import ( "encoding/json" "math" "testing" + "time" "github.com/google/go-cmp/cmp" pilosa "github.com/molecula/featurebase/v3" @@ -305,38 +306,40 @@ func TestFieldInfoMarshal(t *testing.T) { } } -// func TestCheckUnixNanoOverflow(t *testing.T) { -// tests := []struct { -// name string -// epoch time.Time -// wantErr bool -// }{ -// { -// name: "too small", -// epoch: time.Unix(-1, math.MinInt64), -// wantErr: true, -// }, -// { -// name: "just right-1", -// epoch: time.Unix(0, math.MinInt64), -// wantErr: false, -// }, -// { -// name: "just right-2", -// epoch: time.Unix(0, math.MaxInt64), -// wantErr: false, -// }, -// { -// name: "too large", -// epoch: time.Unix(1, math.MaxInt64), -// wantErr: true, -// }, -// } -// for _, tt := range tests { -// t.Run(tt.name, func(t *testing.T) { -// if err := pilosa.CheckEpochNanoOverflow(tt.epoch); (err != nil) != tt.wantErr { -// t.Errorf("checkUnixNanoOverflow() error = %v, wantErr %v", err, tt.wantErr) -// } -// }) -// } -// } +func TestCheckUnixNanoOverflow(t *testing.T) { + minNano = pilosa.MinTimestampNano.UnixNano() + maxNano = pilosa.MaxTimestampNano.UnixNano() + tests := []struct { + name string + epoch time.Time + wantErr bool + }{ + { + name: "too small", + epoch: time.Unix(-1, minNano), + wantErr: true, + }, + { + name: "just right-1", + epoch: time.Unix(0, minNano), + wantErr: false, + }, + { + name: "just right-2", + epoch: time.Unix(0, maxNano), + wantErr: false, + }, + { + name: "too large", + epoch: time.Unix(1, maxNano), + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + 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 97dcf1be6..37518afa5 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/util.go b/util.go index 71ced7b4b..4188c67af 100644 --- a/util.go +++ b/util.go @@ -60,12 +60,6 @@ func GetLoopProgress(start time.Time, now time.Time, iteration uint, total uint) return time.Duration(avgItemTime * float64(itemsLeft)), pctDone } -// FormatTimestampNano returns the string representation of a timestamp given: -// an epoch value, base, and time unit -// func FormatTimestampNano(value, base int64, timeUnit string) string { -// return time.Unix(0, (value+base)*TimeUnitNanos(timeUnit)).UTC().Format(time.RFC3339Nano) // TODO: Samir Should be Dead Code Now -// } - type MemoryUsage struct { Capacity uint64 `json:"capacity"` TotalUse uint64 `json:"totalUsed"` diff --git a/util_test.go b/util_test.go index c1a29bfe3..36eb31ef0 100644 --- a/util_test.go +++ b/util_test.go @@ -72,24 +72,6 @@ func TestGetLoopProgress(t *testing.T) { } } -// func TestFormatTimestampNano(t *testing.T) { -// if FormatTimestampNano(0, 69, "s") != "1970-01-01T00:01:09Z" { -// t.Fatal("Timestamp not formatted properly") -// } -// if FormatTimestampNano(0, 420, "ms") != "1970-01-01T00:00:00.42Z" { -// t.Fatal("Timestamp not formatted properly") -// } -// if FormatTimestampNano(420, 0, "μs") != "1970-01-01T00:00:00.00000042Z" { -// t.Fatal("Timestamp not formatted properly") -// } -// if FormatTimestampNano(420, 69, "us") != "1970-01-01T00:00:00.000489Z" { -// t.Fatal("Timestamp not formatted properly") -// } -// if FormatTimestampNano(69, 420, "ns") != "1970-01-01T00:00:00.000000489Z" { -// t.Fatal("Timestamp not formatted properly") -// } -// } - func TestGetMemoryUsage(t *testing.T) { if _, err := GetMemoryUsage(); err != nil { t.Fatalf("unexpected error getting memory usage: %v", err)