scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix few bugs regarding timestamp

Browse source 
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.
This commit is contained in:
Samir Patel 2022-07-12 13:56:24 -05:00 • committed by Samir Patel
parent 593992312a
commit 33fce8a05b
9 changed files with 435 additions and 244 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
api.go
View file

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

84
api_test.go
View file

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

45
executor.go
View file

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

373
executor_test.go
View file

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

73
field.go
View file

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

73
field_test.go
View file

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

2
fragment.go
View file

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

6
util.go
View file

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

18
util_test.go
View file

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