mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-05 08:10:50 +00:00
Merge branch 'master' into cicd-smoketest
This commit is contained in:
commit
130c2265ca
9 changed files with 228 additions and 59 deletions
|
|
@ -90,21 +90,10 @@ run go tests future:
|
|||
script:
|
||||
- echo "Running featurebase unit tests..."
|
||||
- PKG_LIST=$(go list ./... | grep -v internal/clustertests | paste -s -d, -)
|
||||
- go test -coverprofile=coverage.out -covermode=atomic -coverpkg=${PKG_LIST} ./...
|
||||
- go test -json -coverprofile=coverage.out -covermode=atomic -coverpkg=${PKG_LIST} ./... | tee test-report.out
|
||||
artifacts:
|
||||
paths:
|
||||
- coverage.out
|
||||
|
||||
run go tests with output:
|
||||
stage: test
|
||||
image: golang:$GOVERSION
|
||||
rules:
|
||||
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
|
||||
script:
|
||||
- echo "Running featurebase unit tests to capture JSON output..."
|
||||
- go test -json > test-report.out
|
||||
artifacts:
|
||||
paths:
|
||||
- test-report.out
|
||||
|
||||
upload to sonarcloud:
|
||||
|
|
@ -118,7 +107,6 @@ upload to sonarcloud:
|
|||
- sonar-scanner -Dsonar.projectKey=molecula_featurebase -Dsonar.organization=molecula -Dsonar.sources=. -Dsonar.host.url=https://sonarcloud.io -Dsonar.go.coverage.reportPaths=coverage.out -Dsonar.go.tests.reportPaths=test-report.out -Dsonar.javascript.lcov.reportPaths=lattice/coverage/lcov.info
|
||||
needs:
|
||||
- job: run go tests
|
||||
- job: run go tests with output
|
||||
- job: run jest tests
|
||||
|
||||
build for linux amd64:
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
package client
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
|
|
@ -20,7 +21,9 @@ const (
|
|||
// order. Could be worth sorting everything after translation (as an
|
||||
// option?). Instead of sorting all simultaneously, it might be faster
|
||||
// (more cache friendly) to sort ids and save the swap ops to apply to
|
||||
// everything else that needs to be sorted.
|
||||
// everything else that needs to be sorted. Note: we're already doing
|
||||
// some sorting in importValueData and importMutexData, so if we
|
||||
// implement it at the top level, remember to remove it there.
|
||||
|
||||
// TODO support clearing values? nil values in records are ignored,
|
||||
// but perhaps we could have a special type indicating that a bit or
|
||||
|
|
@ -1216,6 +1219,22 @@ func (b *Batch) makeFragments(frags, clearFrags fragments) (fragments, fragments
|
|||
return frags, clearFrags, nil
|
||||
}
|
||||
|
||||
type valsByIDsSortable struct {
|
||||
ids []uint64
|
||||
vals []int64
|
||||
// shard width so we can compare by shard instead of ID
|
||||
width uint64
|
||||
}
|
||||
|
||||
func (v *valsByIDsSortable) Len() int { return len(v.ids) }
|
||||
|
||||
// comparing on shard rather than ID was twice as fast in informal tests
|
||||
func (v *valsByIDsSortable) Less(i, j int) bool { return v.ids[i]/v.width < v.ids[j]/v.width }
|
||||
func (v *valsByIDsSortable) Swap(i, j int) {
|
||||
v.ids[i], v.ids[j] = v.ids[j], v.ids[i]
|
||||
v.vals[i], v.vals[j] = v.vals[j], v.vals[i]
|
||||
}
|
||||
|
||||
// importValueData imports data for int fields.
|
||||
func (b *Batch) importValueData() error {
|
||||
shardWidth := b.index.ShardWidth()
|
||||
|
|
@ -1246,6 +1265,12 @@ func (b *Batch) importValueData() error {
|
|||
if len(ids) == 0 {
|
||||
continue // TODO test this "all nil" case
|
||||
}
|
||||
|
||||
sc := &valsByIDsSortable{ids: ids, vals: bvalues, width: shardWidth}
|
||||
if !sort.IsSorted(sc) {
|
||||
sort.Sort(sc)
|
||||
}
|
||||
|
||||
curShard := ids[0] / shardWidth
|
||||
startIdx := 0
|
||||
for i := 1; i <= len(ids); i++ {
|
||||
|
|
@ -1285,6 +1310,22 @@ func (b *Batch) importValueData() error {
|
|||
return errors.Wrap(err, "importing value data")
|
||||
}
|
||||
|
||||
type rowsByIDsSortable struct {
|
||||
ids []uint64
|
||||
rows []uint64
|
||||
// shard width so we can compare by shard instead of ID
|
||||
width uint64
|
||||
}
|
||||
|
||||
func (v *rowsByIDsSortable) Len() int { return len(v.ids) }
|
||||
|
||||
// comparing on shard rather than ID was twice as fast in informal tests
|
||||
func (v *rowsByIDsSortable) Less(i, j int) bool { return v.ids[i]/v.width < v.ids[j]/v.width }
|
||||
func (v *rowsByIDsSortable) Swap(i, j int) {
|
||||
v.ids[i], v.ids[j] = v.ids[j], v.ids[i]
|
||||
v.rows[i], v.rows[j] = v.rows[j], v.rows[i]
|
||||
}
|
||||
|
||||
// TODO this should work for bools as well - just need to support them
|
||||
// at batch creation time and when calling Add, I think.
|
||||
func (b *Batch) importMutexData() error {
|
||||
|
|
@ -1319,6 +1360,12 @@ func (b *Batch) importMutexData() error {
|
|||
if len(ids) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
sc := &rowsByIDsSortable{ids: ids, rows: rowIDs, width: shardWidth}
|
||||
if !sort.IsSorted(sc) {
|
||||
sort.Sort(sc)
|
||||
}
|
||||
|
||||
curShard := ids[0] / shardWidth
|
||||
startIdx := 0
|
||||
for i := 1; i <= len(ids); i++ {
|
||||
|
|
|
|||
|
|
@ -1,23 +1,48 @@
|
|||
// Copyright 2021 Molecula Corp. All rights reserved.
|
||||
//go:build integration
|
||||
// +build integration
|
||||
|
||||
package client
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/molecula/featurebase/v2/test"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
func TestStringSliceCombos(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func NewTestClient(t *testing.T, c *test.Cluster) *Client {
|
||||
client, err := NewClient(c.Nodes[0].URL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
func TestAgainstCluster(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
client := NewTestClient(t, c)
|
||||
|
||||
t.Run("string-slice-combos", func(t *testing.T) { testStringSliceCombos(t, c, client) })
|
||||
t.Run("import-batch-ints", func(t *testing.T) { testImportBatchInts(t, c, client) })
|
||||
t.Run("import-batch-sorting", func(t *testing.T) { testImportBatchSorting(t, c, client) })
|
||||
t.Run("test-trim-null", func(t *testing.T) { testTrimNull(t, c, client) })
|
||||
t.Run("test-string-slice-empty-and-nil", func(t *testing.T) { testStringSliceEmptyAndNil(t, c, client) })
|
||||
t.Run("test-string-slice", func(t *testing.T) { testStringSlice(t, c, client) })
|
||||
t.Run("test-single-clear-batch-regression", func(t *testing.T) { testSingleClearBatchRegression(t, c, client) })
|
||||
t.Run("test-batches", func(t *testing.T) { testBatches(t, c, client) })
|
||||
t.Run("batches-strings-ids", func(t *testing.T) { testBatchesStringIDs(t, c, client) })
|
||||
t.Run("test-batch-staleness", func(t *testing.T) { testBatchStaleness(t, c, client) })
|
||||
}
|
||||
|
||||
func testStringSliceCombos(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("test-string-slicecombos")
|
||||
idx := schema.Index("test-string-slice-combos")
|
||||
fields := make([]*Field, 1)
|
||||
fields[0] = idx.Field("a1", OptFieldKeys(true), OptFieldTypeSet(CacheTypeRanked, 100))
|
||||
err := client.SyncSchema(schema)
|
||||
|
|
@ -152,10 +177,9 @@ func ingestRecords(records []Row, batch *Batch) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func TestImportBatchInts(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testImportBatchInts(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-blah")
|
||||
idx := schema.Index("test-import-batch-ints")
|
||||
field := idx.Field("anint", OptFieldTypeInt())
|
||||
err := client.SyncSchema(schema)
|
||||
if err != nil {
|
||||
|
|
@ -212,10 +236,59 @@ func TestImportBatchInts(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestTrimNull(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testImportBatchSorting(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-null")
|
||||
idx := schema.Index("test-import-batch-sorting")
|
||||
field := idx.Field("anint", OptFieldTypeInt())
|
||||
field2 := idx.Field("amutex", OptFieldTypeMutex(CacheTypeNone, 0))
|
||||
err := client.SyncSchema(schema)
|
||||
if err != nil {
|
||||
t.Fatalf("syncing schema: %v", err)
|
||||
}
|
||||
|
||||
b, err := NewBatch(client, 100, idx, []*Field{field, field2})
|
||||
if err != nil {
|
||||
t.Fatalf("getting batch: %v", err)
|
||||
}
|
||||
|
||||
r := Row{Values: make([]interface{}, 2)}
|
||||
|
||||
rnd := rand.New(rand.NewSource(7))
|
||||
|
||||
// generate 100 records randomly spread/ordered across multiple
|
||||
// shards to test sorting on int/mutex fields
|
||||
for i := 0; i < 100; i++ {
|
||||
id := rnd.Intn(10_000_000)
|
||||
r.ID = uint64(id)
|
||||
r.Values[0] = int64(id)
|
||||
r.Values[1] = uint64(id)
|
||||
err := b.Add(r)
|
||||
if err != nil && err != ErrBatchNowFull {
|
||||
t.Fatalf("adding to batch: %v", err)
|
||||
}
|
||||
}
|
||||
err = b.Import()
|
||||
if err != nil {
|
||||
t.Fatalf("importing: %v", err)
|
||||
}
|
||||
|
||||
err = b.Import()
|
||||
if err != nil {
|
||||
t.Fatalf("second import: %v", err)
|
||||
}
|
||||
|
||||
resp, err := client.Query(idx.RawQuery("Count(All())"))
|
||||
if err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
}
|
||||
if res := resp.Results()[0]; res.Count() != 100 {
|
||||
t.Fatalf("unexpected result: %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func testTrimNull(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("test-trim-null")
|
||||
field := idx.Field("empty", OptFieldTypeInt())
|
||||
err := client.SyncSchema(schema)
|
||||
if err != nil {
|
||||
|
|
@ -299,8 +372,7 @@ func TestTrimNull(t *testing.T) {
|
|||
|
||||
}
|
||||
|
||||
func TestStringSliceEmptyAndNil(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testStringSliceEmptyAndNil(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("test-string-slice-nil")
|
||||
fields := make([]*Field, 1)
|
||||
|
|
@ -397,8 +469,7 @@ func TestStringSliceEmptyAndNil(t *testing.T) {
|
|||
|
||||
}
|
||||
|
||||
func TestStringSlice(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testStringSlice(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("test-string-slice")
|
||||
fields := make([]*Field, 1)
|
||||
|
|
@ -513,10 +584,9 @@ func TestStringSlice(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestSingleClearBatchRegression(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testSingleClearBatchRegression(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-blah")
|
||||
idx := schema.Index("test-single-clear-batch-regression")
|
||||
numFields := 1
|
||||
fields := make([]*Field, numFields)
|
||||
fields[0] = idx.Field("zero", OptFieldKeys(true))
|
||||
|
|
@ -565,10 +635,9 @@ func TestSingleClearBatchRegression(t *testing.T) {
|
|||
|
||||
}
|
||||
|
||||
func TestBatches(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testBatches(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-blah")
|
||||
idx := schema.Index("test-batches")
|
||||
numFields := 5
|
||||
fields := make([]*Field, numFields)
|
||||
fields[0] = idx.Field("zero", OptFieldKeys(true))
|
||||
|
|
@ -890,8 +959,8 @@ func TestBatches(t *testing.T) {
|
|||
}
|
||||
}
|
||||
res := results[1]
|
||||
cols := res.Row().Columns
|
||||
if !reflect.DeepEqual(cols, []uint64{0, 2, 4, 6, 8, 10, 12, 14, 16, 18}) {
|
||||
|
||||
if cols := res.Row().Columns; !reflect.DeepEqual(cols, []uint64{0, 2, 4, 6, 8, 10, 12, 14, 16, 18}) {
|
||||
t.Fatalf("unexpected columns for field 1 row b: %v", cols)
|
||||
}
|
||||
|
||||
|
|
@ -919,23 +988,25 @@ func TestBatches(t *testing.T) {
|
|||
t.Fatalf("querying: %v", err)
|
||||
}
|
||||
results = resp.Results()
|
||||
cols = results[0].Row().Columns
|
||||
if !reflect.DeepEqual(cols, []uint64{0, 1, 2, 3, 4, 5, 6, 7, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28}) {
|
||||
|
||||
if cols := results[0].Row().Columns; !reflect.DeepEqual(cols, []uint64{0, 1, 2, 3, 4, 5, 6, 7, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28}) {
|
||||
t.Fatalf("all columns (but 8) should be greater than -11, but got: %v", cols)
|
||||
}
|
||||
cols = results[1].Row().Columns
|
||||
if !reflect.DeepEqual(cols, []uint64{19, 21, 23, 25, 27}) {
|
||||
|
||||
if cols := results[1].Row().Columns; !reflect.DeepEqual(cols, []uint64{19, 21, 23, 25, 27}) {
|
||||
t.Fatalf("wrong cols for ==0: %v", cols)
|
||||
}
|
||||
cols = results[2].Row().Columns
|
||||
if !reflect.DeepEqual(cols, []uint64{20, 22, 24, 26, 28}) {
|
||||
|
||||
if cols := results[2].Row().Columns; !reflect.DeepEqual(cols, []uint64{20, 22, 24, 26, 28}) {
|
||||
t.Fatalf("wrong cols for ==100: %v", cols)
|
||||
}
|
||||
cols = results[3].Row().Columns
|
||||
|
||||
cols := results[3].Row().Columns
|
||||
exp := []uint64{0, 2, 4, 6, 10, 12, 14, 16, 18}
|
||||
if !reflect.DeepEqual(cols, exp) {
|
||||
t.Fatalf("wrong cols for January: got/want\n%v\n%v", cols, exp)
|
||||
}
|
||||
|
||||
cols = results[4].Row().Columns
|
||||
exp = []uint64{1, 3, 5, 7}
|
||||
if !reflect.DeepEqual(cols, exp) {
|
||||
|
|
@ -977,10 +1048,9 @@ func TestBatches(t *testing.T) {
|
|||
// TODO test importing across multiple shards
|
||||
}
|
||||
|
||||
func TestBatchesStringIDs(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testBatchesStringIDs(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-blah", OptIndexKeys(true))
|
||||
idx := schema.Index("batches-strings-ids", OptIndexKeys(true))
|
||||
fields := make([]*Field, 3)
|
||||
fields[0] = idx.Field("zero", OptFieldKeys(true))
|
||||
fields[1] = idx.Field("one", OptFieldTypeMutex(CacheTypeNone, 0), OptFieldKeys(true))
|
||||
|
|
@ -1263,10 +1333,9 @@ func TestQuantizedTime(t *testing.T) {
|
|||
|
||||
}
|
||||
|
||||
func TestBatchStaleness(t *testing.T) {
|
||||
client := DefaultClient()
|
||||
func testBatchStaleness(t *testing.T, c *test.Cluster, client *Client) {
|
||||
schema := NewSchema()
|
||||
idx := schema.Index("gopilosatest-blah")
|
||||
idx := schema.Index("test-batch-staleness")
|
||||
field := idx.Field("anint", OptFieldTypeInt())
|
||||
err := client.SyncSchema(schema)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -269,37 +269,37 @@ func TestIngestAPIBatch(t *testing.T) {
|
|||
if resp, err := cli.Query(NewPQLBaseQuery("Row(bint==-2)", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(bint==-2) result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
if resp, err := cli.Query(NewPQLBaseQuery("Row(cid=9)", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(cid=9) result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
if resp, err := cli.Query(NewPQLBaseQuery("Row(dtimestamp=='2010-10-18T02:07:03Z')", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(dtimestamp=='2010-10-18T02:07:03Z') result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
if resp, err := cli.Query(NewPQLBaseQuery("Row(etime=e, from='2010-01-01', to='2010-01-02')", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(etime=e, from='2010-01-01', to='2010-01-02') result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
if resp, err := cli.Query(NewPQLBaseQuery("Row(fdecimal==1.234)", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(fdecimal==1.234) result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
if resp, err := cli.Query(NewPQLBaseQuery("Row(gbool=true)", &Index{name: "test-1", options: &IndexOptions{}}, nil)); err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
} else if len(resp.Result().Row().Columns) != 1 || resp.Result().Row().Columns[0] != uint64(7) {
|
||||
t.Fatalf("unexpected Row(asr=a) result: %+v", resp.Result().Row().Columns)
|
||||
t.Fatalf("unexpected Row(gbool=true) result: %+v", resp.Result().Row().Columns)
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
|
||||
pilosa "github.com/molecula/featurebase/v2"
|
||||
fb_http "github.com/molecula/featurebase/v2/http"
|
||||
"github.com/molecula/featurebase/v2/logger"
|
||||
"github.com/molecula/featurebase/v2/server"
|
||||
"github.com/molecula/featurebase/v2/topology"
|
||||
"github.com/pkg/errors"
|
||||
|
|
@ -203,6 +204,8 @@ func (cmd *RestoreCommand) newClient() *retryablehttp.Client {
|
|||
client.RetryWaitMin = min
|
||||
client.RetryMax = int(attempts)
|
||||
client.CheckRetry = retryWith400
|
||||
client.Logger = logger.NopLogger
|
||||
|
||||
return client
|
||||
}
|
||||
|
||||
|
|
|
|||
18
executor.go
18
executor.go
|
|
@ -1181,6 +1181,8 @@ func (e *executor) executeDistinct(ctx context.Context, qcx *Qcx, index string,
|
|||
return other.Union(v.(*Row))
|
||||
case nil:
|
||||
return v
|
||||
case DistinctTimestamp:
|
||||
return other.Union(v.(DistinctTimestamp))
|
||||
default:
|
||||
return errors.Errorf("unexpected return type from executeDistinctShard: %+v %T", other, other)
|
||||
}
|
||||
|
|
@ -1633,6 +1635,22 @@ type DistinctTimestamp struct {
|
|||
Name string
|
||||
}
|
||||
|
||||
// Union returns the union of the values of `d` and `other`
|
||||
func (d *DistinctTimestamp) Union(other DistinctTimestamp) DistinctTimestamp {
|
||||
both := map[string]string{}
|
||||
for _, val := range d.Values {
|
||||
both[val] = val
|
||||
}
|
||||
for _, val := range other.Values {
|
||||
both[val] = val
|
||||
}
|
||||
vals := []string{}
|
||||
for key := range both {
|
||||
vals = append(vals, key)
|
||||
}
|
||||
return DistinctTimestamp{Name: d.Name, Values: vals}
|
||||
}
|
||||
|
||||
func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, filterBitmap *roaring.Bitmap) (result *Row, err0 error) {
|
||||
index := idx.Name()
|
||||
tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})
|
||||
|
|
|
|||
|
|
@ -504,3 +504,44 @@ func TestGetScaledInt(t *testing.T) {
|
|||
}
|
||||
|
||||
}
|
||||
|
||||
func TestDistinctTimestampUnion(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
a DistinctTimestamp
|
||||
b DistinctTimestamp
|
||||
expected DistinctTimestamp
|
||||
}{
|
||||
{
|
||||
name: "empty other",
|
||||
a: DistinctTimestamp{Name: "a", Values: []string{"a", "b", "c"}},
|
||||
b: DistinctTimestamp{Name: "a", Values: []string{}},
|
||||
expected: DistinctTimestamp{Name: "a", Values: []string{"a", "b", "c"}},
|
||||
},
|
||||
{
|
||||
name: "one more in other",
|
||||
a: DistinctTimestamp{Name: "a", Values: []string{"a", "b", "c"}},
|
||||
b: DistinctTimestamp{Name: "a", Values: []string{"a", "b", "c", "d"}},
|
||||
expected: DistinctTimestamp{Name: "a", Values: []string{"a", "b", "c", "d"}},
|
||||
},
|
||||
}
|
||||
for _, test := range cases {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
res := test.a.Union(test.b)
|
||||
allThere := true
|
||||
for _, val := range res.Values {
|
||||
here := false
|
||||
for _, expected := range test.expected.Values {
|
||||
if val == expected {
|
||||
here = true
|
||||
break
|
||||
}
|
||||
}
|
||||
allThere = allThere && here
|
||||
}
|
||||
if !allThere {
|
||||
t.Errorf("expected %v, got %v", test.expected, res)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6752,9 +6752,10 @@ func variousQueriesCountDistinctTimestamp(t *testing.T, c *test.Cluster) {
|
|||
c.CreateField(t, index, pilosa.IndexOptions{TrackExistence: true}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s"))
|
||||
|
||||
// add some data
|
||||
data := []string{"2010-01-02T12:32:00Z", "2010-04-20T12:32:00Z", "2011-04-20T12:32:00Z"}
|
||||
data := []string{"2010-01-02T12:32:00Z", "2010-04-20T12:32:00Z", "2011-04-20T12:59:00Z", "2011-04-20T12:40:00Z", "2011-04-20T12:32:00Z"}
|
||||
|
||||
for i, datum := range data {
|
||||
c.Query(t, index, fmt.Sprintf("Set(%d, ts=\"%s\")", i+10, datum))
|
||||
c.Query(t, index, fmt.Sprintf("Set(%d, ts=\"%s\")", i*(1<<20), datum))
|
||||
}
|
||||
|
||||
// query the Count of Distinct vals in field ts
|
||||
|
|
|
|||
|
|
@ -75,13 +75,14 @@ func WithClientRetryPeriod(period time.Duration) InternalClientOption {
|
|||
if attempts < 1 {
|
||||
attempts = 1
|
||||
}
|
||||
fmt.Println("attempts: ", int(attempts))
|
||||
|
||||
return func(c *InternalClient) {
|
||||
rc := retryablehttp.NewClient()
|
||||
rc.HTTPClient = c.httpClient
|
||||
rc.RetryWaitMin = min
|
||||
rc.RetryMax = int(attempts)
|
||||
rc.CheckRetry = retryWith400Policy
|
||||
rc.Logger = logger.NopLogger
|
||||
c.retryableClient = rc
|
||||
}
|
||||
}
|
||||
|
|
@ -123,6 +124,7 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o
|
|||
rc := retryablehttp.NewClient()
|
||||
rc.HTTPClient = ic.httpClient
|
||||
rc.CheckRetry = noRetryPolicy
|
||||
rc.Logger = logger.NopLogger
|
||||
ic.retryableClient = rc
|
||||
}
|
||||
return ic
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue