scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 16:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
featurebase/http/client_test.go
Matt Jaffee 5dcabfcc7f
support for decimal fields 
This commit adds a Decimal field type which is implemented mostly with
the Int field. It adds an optional "Scale" value to the Int field
which means that the values stored in that field are actually meant to
be divided by 10^Scale before being interpreted.

In order to make use of this functionality, we extend the importValue
request to allow a slice of floats rather than just int64. If the
slice of floats is present, each float in the slice is multiplied by
10^Scale and converted to an int64 before being imported. If a slice
of int64 is imported to a Decimal field, it is treated normally, and
scale is ignored. This allows the conversion to be handled at the
client side if desired.

Currently there are Field level methods for querying Float values out
of a decimal field, but no support in PQL or the executor for getting
float values. Going to wait until I can use the generic result type
before doing that, so for now, any values queried will be the scaled
integer values.

needed to add client support for importing float values, and did this
by adding a more general and simplified client method for value
imports.

rewrote api.ImportValue to use the new method which should be more
performant and efficient.

allow floats to be "pilosa import"ed into decimal fields
2019-10-29 16:36:14 -05:00

1064 lines
32 KiB
Go
Raw Blame History

// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package http_test
import (
"bufio"
"bytes"
"context"
"encoding/hex"
"fmt"
gohttp "net/http"
"reflect"
"testing"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
)
// Test distributed TopN Row count across 3 nodes.
func TestClient_MultiNode(t *testing.T) {
c := test.MustRunCluster(t, 3,
[]server.CommandOption{
server.OptCommandServerOptions(pilosa.OptServerNodeID("node0"), pilosa.OptServerClusterHasher(&test.ModHasher{}))},
[]server.CommandOption{
server.OptCommandServerOptions(pilosa.OptServerNodeID("node1"), pilosa.OptServerClusterHasher(&test.ModHasher{}))},
[]server.CommandOption{
server.OptCommandServerOptions(pilosa.OptServerNodeID("node2"), pilosa.OptServerClusterHasher(&test.ModHasher{}))},
)
defer c.Close()
hldr := []test.Holder{}
for _, command := range c {
hldr = append(hldr, test.Holder{Holder: command.Server.Holder()})
}
// Create a dispersed set of bitmaps across 3 nodes such that each individual node and shard width increment would reveal a different TopN.
shardNums := []uint64{1, 2, 6}
// This was generated with: `owns := s[i].Handler.Handler.API.Cluster.OwnsShards("i", 20, s[i].HostURI())`
owns := [][]uint64{
{1, 3, 4, 8, 10, 13, 17, 19},
{2, 5, 7, 11, 12, 14, 18},
{0, 6, 9, 15, 16, 20},
}
for i, num := range shardNums {
ownsNum := false
for _, ownNum := range owns[i] {
if ownNum == num {
ownsNum = true
break
}
}
if !ownsNum {
t.Fatalf("Trying to use shard %d on host %s, but it doesn't own that shard. It owns %v", num, c[i].URL(), owns)
}
}
baseBit0 := pilosa.ShardWidth * shardNums[0]
baseBit1 := pilosa.ShardWidth * shardNums[1]
baseBit2 := pilosa.ShardWidth * shardNums[2]
maxShard := uint64(0)
for _, x := range shardNums {
if x > maxShard {
maxShard = x
}
}
_, err := c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
hldr[0].MustSetBits("i", "f", 100, baseBit0+10)
hldr[0].MustSetBits("i", "f", 4, baseBit0+10, baseBit0+11, baseBit0+12)
hldr[0].MustSetBits("i", "f", 4, baseBit0+10, baseBit0+11, baseBit0+12, baseBit0+13, baseBit0+14, baseBit0+15)
hldr[0].MustSetBits("i", "f", 2, baseBit0+1, baseBit0+2, baseBit0+3, baseBit0+4)
hldr[0].MustSetBits("i", "f", 3, baseBit0+1, baseBit0+2, baseBit0+3, baseBit0+4, baseBit0+5)
hldr[0].MustSetBits("i", "f", 22, baseBit0+1, baseBit0+2)
hldr[1].MustSetBits("i", "f", 99, baseBit1+1, baseBit1+2, baseBit1+3, baseBit1+4)
hldr[1].MustSetBits("i", "f", 100, baseBit1+1, baseBit1+2, baseBit1+3, baseBit1+4, baseBit1+5, baseBit1+6, baseBit1+7, baseBit1+8, baseBit1+9, baseBit1+10)
hldr[1].MustSetBits("i", "f", 98, baseBit1+1, baseBit1+2, baseBit1+3, baseBit1+4, baseBit1+5, baseBit1+6)
hldr[1].MustSetBits("i", "f", 1, baseBit1+4)
hldr[1].MustSetBits("i", "f", 22, baseBit1+1, baseBit1+2, baseBit1+3, baseBit1+4, baseBit1+5)
hldr[2].MustSetBits("i", "f", 24, baseBit2+10, baseBit2+11, baseBit2+12, baseBit2+13, baseBit2+14)
hldr[2].MustSetBits("i", "f", 20, baseBit2+10, baseBit2+11, baseBit2+12, baseBit2+13)
hldr[2].MustSetBits("i", "f", 21, baseBit2+10)
hldr[2].MustSetBits("i", "f", 100, baseBit2+10)
hldr[2].MustSetBits("i", "f", 99, baseBit2+10, baseBit2+11, baseBit2+12)
hldr[2].MustSetBits("i", "f", 98, baseBit2+10, baseBit2+11)
hldr[2].MustSetBits("i", "f", 22, baseBit2+10, baseBit2+11, baseBit2+12)
// Rebuild the RankCache.
// We have to do this to avoid the 10-second cache invalidation delay
// built into cache.Invalidate()
err = c[0].RecalculateCaches()
if err != nil {
t.Fatalf("recalculating cache: %v", err)
}
err = c[1].RecalculateCaches()
if err != nil {
t.Fatalf("recalculating cache: %v", err)
}
err = c[2].RecalculateCaches()
if err != nil {
t.Fatalf("recalculating cache: %v", err)
}
// Connect to each node to compare results.
client := make([]*Client, 3)
client[0] = MustNewClient(c[0].URL(), http.GetHTTPClient(nil))
client[1] = MustNewClient(c[1].URL(), http.GetHTTPClient(nil))
client[2] = MustNewClient(c[2].URL(), http.GetHTTPClient(nil))
topN := 4
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`TopN(f, n=%d)`, topN),
Remote: false,
}
result, err := client[0].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
// Test must return exactly N results.
if len(result.Results[0].([]pilosa.Pair)) != topN {
t.Fatalf("unexpected number of TopN results: %s", spew.Sdump(result))
}
p := []pilosa.Pair{
{ID: 100, Count: 12},
{ID: 22, Count: 10},
{ID: 98, Count: 8},
{ID: 99, Count: 7}}
// Valdidate the Top 4 result counts.
if !reflect.DeepEqual(result.Results[0].([]pilosa.Pair), p) {
t.Fatalf("Invalid TopN result set: %s", spew.Sdump(result))
}
result1, err := client[1].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
result2, err := client[2].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
// Compare TopN results across all nodes in the cluster.
if !reflect.DeepEqual(result, result1) {
t.Fatalf("TopN result should be the same on node0 and node1: %s", spew.Sdump(result1))
}
if !reflect.DeepEqual(result, result2) {
t.Fatalf("TopN result should be the same on node0 and node2: %s", spew.Sdump(result2))
}
}
// Ensure client can export data.
func TestClient_Export(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false})
cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "unkeyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
c := MustNewClient(host, http.GetHTTPClient(nil))
data := []pilosa.Bit{
{RowID: 1, ColumnID: 100, RowKey: "row1", ColumnKey: "col100"},
{RowID: 1, ColumnID: 101, RowKey: "row1", ColumnKey: "col101"},
{RowID: 1, ColumnID: 102, RowKey: "row1", ColumnKey: "col102"},
{RowID: 1, ColumnID: 103, RowKey: "row1", ColumnKey: "col103"},
{RowID: 2, ColumnID: 200, RowKey: "row2", ColumnKey: "col200"},
{RowID: 2, ColumnID: 201, RowKey: "row2", ColumnKey: "col201"},
{RowID: 2, ColumnID: 202, RowKey: "row2", ColumnKey: "col202"},
{RowID: 2, ColumnID: 203, RowKey: "row2", ColumnKey: "col203"},
}
t.Run("Export unkeyed,unkeyedf", func(t *testing.T) {
// Populate data.
for _, bit := range data {
_, err := c.Query(context.Background(), "unkeyed", &pilosa.QueryRequest{
Query: fmt.Sprintf(`Set(%d, unkeyedf=%d)`, bit.ColumnID, bit.RowID),
Remote: false,
})
if err != nil {
t.Fatal(err)
}
}
buf := bytes.NewBuffer(nil)
bw := bufio.NewWriter(buf)
// Send export request.
if err := c.ExportCSV(context.Background(), "unkeyed", "unkeyedf", 0, bw); err != nil {
t.Fatal(err)
}
got := buf.String()
// Expected output.
exp := ""
for _, bit := range data {
exp += fmt.Sprintf("%d,%d\n", bit.RowID, bit.ColumnID)
}
// Verify data.
if got != exp {
t.Fatalf("unexpected export data: %s", got)
}
})
t.Run("Export unkeyed,keyedf", func(t *testing.T) {
// Populate data.
for _, bit := range data {
_, err := c.Query(context.Background(), "unkeyed", &pilosa.QueryRequest{
Query: fmt.Sprintf(`Set(%d, keyedf=%s)`, bit.ColumnID, bit.RowKey),
Remote: false,
})
if err != nil {
t.Fatal(err)
}
}
buf := bytes.NewBuffer(nil)
bw := bufio.NewWriter(buf)
// Send export request.
if err := c.ExportCSV(context.Background(), "unkeyed", "keyedf", 0, bw); err != nil {
t.Fatal(err)
}
got := buf.String()
// Expected output.
exp := ""
for _, bit := range data {
exp += fmt.Sprintf("%s,%d\n", bit.RowKey, bit.ColumnID)
}
// Verify data.
if got != exp {
t.Fatalf("unexpected export data: %s", got)
}
})
t.Run("Export keyed,unkeyedf", func(t *testing.T) {
// Populate data.
for _, bit := range data {
_, err := c.Query(context.Background(), "keyed", &pilosa.QueryRequest{
Query: fmt.Sprintf(`Set("%s", unkeyedf=%d)`, bit.ColumnKey, bit.RowID),
Remote: false,
})
if err != nil {
t.Fatal(err)
}
}
buf := bytes.NewBuffer(nil)
bw := bufio.NewWriter(buf)
// Send export request.
if err := c.ExportCSV(context.Background(), "keyed", "unkeyedf", 0, bw); err != nil {
t.Fatal(err)
}
got := buf.String()
// Expected output.
exp := ""
for _, bit := range data {
exp += fmt.Sprintf("%d,%s\n", bit.RowID, bit.ColumnKey)
}
// Verify data.
if got != exp {
t.Fatalf("unexpected export data: %s", got)
}
})
t.Run("Export keyed,keyedf", func(t *testing.T) {
// Populate data.
for _, bit := range data {
_, err := c.Query(context.Background(), "keyed", &pilosa.QueryRequest{
Query: fmt.Sprintf(`Set("%s", keyedf=%s)`, bit.ColumnKey, bit.RowKey),
Remote: false,
})
if err != nil {
t.Fatal(err)
}
}
buf := bytes.NewBuffer(nil)
bw := bufio.NewWriter(buf)
// Send export request.
if err := c.ExportCSV(context.Background(), "keyed", "keyedf", 0, bw); err != nil {
t.Fatal(err)
}
got := buf.String()
// Expected output.
exp := ""
for _, bit := range data {
exp += fmt.Sprintf("%s,%s\n", bit.RowKey, bit.ColumnKey)
}
// Verify data.
if got != exp {
t.Fatalf("unexpected export data: %s", got)
}
})
}
// Ensure client can bulk import data.
func TestClient_Import(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
// Load bitmap into cache to ensure cache gets updated.
hldr.SetBit("i", "f", 1, 0) // set a bit so the view gets created.
hldr.Row("i", "f", 0)
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{
{RowID: 0, ColumnID: 1},
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}); err != nil {
t.Fatal(err)
}
// Verify data.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 5}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{6}) {
t.Fatalf("unexpected columns: %+v", a)
}
// Clear some data.
if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify data.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected columns: %+v", a)
}
}
// Ensure client can bulk import data.
func TestClient_ImportRoaring(t *testing.T) {
cluster := test.MustNewCluster(t, 2)
for _, c := range cluster {
c.Config.Cluster.ReplicaN = 2
}
err := cluster.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
defer cluster.Close()
_, err = cluster[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = cluster[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 100))
if err != nil {
t.Fatalf("creating field: %v", err)
}
_, err = cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Set(0, f=1)"})
if err != nil {
t.Fatalf("querying: %v", err)
}
// Send import request.
host := cluster[0].URL()
c := MustNewClient(host, http.GetHTTPClient(nil))
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq := makeImportRoaringRequest(false, "3B3001000100000900010000000100010009000100")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
hldr := test.Holder{Holder: cluster[0].Server.Holder()}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected columns: %+v", a)
}
hldr2 := test.Holder{Holder: cluster[1].Server.Holder()}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
// [65539, 65540]
roaringReq = makeImportRoaringRequest(true, "3A30000001000000010001001000000003000400")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
// [4, 6, 65537, 65539]
roaringReq = makeImportRoaringRequest(true, "3A300000020000000000010001000100180000001C0000000400060001000300")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 5, 7, 8, 9, 10}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3, 5, 7, 8, 9, 10}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Ensure that sending a roaring import with the clear flag works as expected.
// [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 65537]
roaringReq = makeImportRoaringRequest(true, "3B3001000100000900010000000100010009000100")
if err := c.ImportRoaring(context.Background(), &cluster[0].API.Node().URI, "i", "f", 0, false, roaringReq); err != nil {
t.Fatal(err)
}
// Verify data on node 0.
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
// Verify data on node 1.
if a := hldr2.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
if a := hldr2.Row("i", "f", 1).Columns(); !reflect.DeepEqual(a, []uint64{0}) {
t.Fatalf("unexpected clear columns: %+v", a)
}
}
// Ensure client can bulk import data.
func TestClient_ImportKeys(t *testing.T) {
t.Run("SingleNode", func(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false})
cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
t.Run("Import keyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
t.Run("Import keyed,unkeyedf", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "unkeyedf", 0, []pilosa.Bit{
{RowID: 1, ColumnKey: "eve"},
{RowID: 1, ColumnKey: "alice"},
{RowID: 1, ColumnKey: "bob"},
{RowID: 2, ColumnKey: "eve"},
{RowID: 2, ColumnKey: "alice"},
{RowID: 3, ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(unkeyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{ID: 1, Count: 3},
{ID: 2, Count: 2},
{ID: 3, Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
t.Run("Import unkeyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "unkeyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnID: 1},
{RowKey: "green", ColumnID: 2},
{RowKey: "green", ColumnID: 3},
{RowKey: "blue", ColumnID: 1},
{RowKey: "blue", ColumnID: 2},
{RowKey: "purple", ColumnID: 1},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "unkeyed",
Query: "TopN(keyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
})
t.Run("MultiNode", func(t *testing.T) {
cluster := test.MustRunCluster(t, 2)
defer cluster.Close()
cmd0 := cluster[0]
cmd1 := cluster[1]
host0 := cmd0.URL()
host1 := cmd1.URL()
cmd0.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd0.MustCreateField(t, "keyed", "keyedf0", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd0.MustCreateField(t, "keyed", "keyedf1", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c0 := MustNewClient(host0, http.GetHTTPClient(nil))
c1 := MustNewClient(host1, http.GetHTTPClient(nil))
// Import to node0.
t.Run("Import node0", func(t *testing.T) {
if err := c0.ImportK(context.Background(), "keyed", "keyedf0", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd0.MustRecalculateCaches(t)
resp := cmd0.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf0)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
// Import to node1 (ensure import is routed to coordinator for translation).
t.Run("Import node1", func(t *testing.T) {
if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
// Wait for translation replication.
time.Sleep(500 * time.Millisecond)
cmd1.MustRecalculateCaches(t)
resp := cmd1.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf1)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %#v", pairs)
}
})
})
t.Run("IntegerFieldSingleNode", func(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
fldName := "f"
// Load bitmap into cache to ensure cache gets updated.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
field, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-100, 100))
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col1", Value: -10},
{ColumnKey: "col2", Value: 20},
{ColumnKey: "col3", Value: 40},
}); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err := field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 50 || cnt != 3 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt)
}
// Verify range.
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`Row(%s>10)`, fldName),
Remote: false,
}
result, err := c.Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col2", "col3"}) {
t.Fatalf("unexpected column keys: %s", spew.Sdump(result))
}
// Clear data.
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col2", Value: 20},
}, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err = field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 30 || cnt != 2 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=30, cnt=2", sum, cnt)
}
// Verify Range.
queryRequest = &pilosa.QueryRequest{
Query: fmt.Sprintf(`Row(%s>10)`, fldName),
Remote: false,
}
result, err = c.Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col3"}) {
t.Fatalf("unexpected column keys: %s", spew.Sdump(result))
}
})
}
// Ensure client can bulk import value data.
func TestClient_ImportValue(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
fldName := "f"
// Load bitmap into cache to ensure cache gets updated.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
field, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-100, 100))
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 2, Value: 20},
{ColumnID: 3, Value: 40},
}); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err := field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 50 || cnt != 3 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt)
}
// Verify Min.
min, cnt, err := field.Min(nil, fldName)
if err != nil {
t.Fatal(err)
}
if min != -10 || cnt != 1 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=-10, cnt=1", min, cnt)
}
// Verify Min with Filter.
filter, err := field.Range(fldName, pql.GT, 40)
if err != nil {
t.Fatal(err)
}
min, cnt, err = field.Min(filter, fldName)
if err != nil {
t.Fatal(err)
}
if min != 0 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=0, cnt=0", min, cnt)
}
// Verify Max.
max, cnt, err := field.Max(nil, fldName)
if err != nil {
t.Fatal(err)
}
if max != 40 || cnt != 1 {
t.Fatalf("unexpected values: got max=%v, count=%v; expected max=40, cnt=1", max, cnt)
}
// Send import request.
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 3, Value: 40},
}, pilosa.OptImportOptionsClear(true)); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err = field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 20 || cnt != 1 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=20, cnt=1", sum, cnt)
}
// Verify Min with Filter.
filter, err = field.Range(fldName, pql.GT, 40)
if err != nil {
t.Fatal(err)
}
min, cnt, err = field.Min(filter, fldName)
if err != nil {
t.Fatal(err)
}
if min != 0 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=0, cnt=0", min, cnt)
}
// Verify Max.
max, cnt, err = field.Max(nil, fldName)
if err != nil {
t.Fatal(err)
}
if max != 20 || cnt != 1 {
t.Fatalf("unexpected values: got max=%v, count=%v; expected max=20, cnt=1", max, cnt)
}
}
// Ensure client can bulk import data while tracking existence.
func TestClient_ImportExistence(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
t.Run("Set", func(t *testing.T) {
idxName := "iset"
fldName := "fset"
index := hldr.MustCreateIndexIfNotExists(idxName, pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateFieldIfNotExists(fldName)
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.Import(context.Background(), idxName, fldName, 0, []pilosa.Bit{
{RowID: 0, ColumnID: 1},
{RowID: 0, ColumnID: 5},
{RowID: 200, ColumnID: 6},
}); err != nil {
t.Fatal(err)
}
// Verify data.
if a := hldr.Row(idxName, fldName, 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 5}) {
t.Fatalf("unexpected columns: %+v", a)
}
if a := hldr.Row(idxName, fldName, 200).Columns(); !reflect.DeepEqual(a, []uint64{6}) {
t.Fatalf("unexpected columns: %+v", a)
}
// Verify existence.
if a := hldr.ReadRow(idxName, "_exists", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 5, 6}) {
t.Fatalf("unexpected existence columns: %+v", a)
}
})
t.Run("Int", func(t *testing.T) {
idxName := "iint"
fldName := "fint"
index := hldr.MustCreateIndexIfNotExists(idxName, pilosa.IndexOptions{TrackExistence: true})
field, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-100, 100))
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), idxName, fldName, 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 2, Value: 20},
{ColumnID: 3, Value: 40},
}); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err := field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 50 || cnt != 3 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt)
}
// Verify existence.
if a := hldr.ReadRow(idxName, "_exists", 0).Columns(); !reflect.DeepEqual(a, []uint64{1, 2, 3}) {
t.Fatalf("unexpected existence columns: %+v", a)
}
})
}
// Ensure client can retrieve a list of all checksums for blocks in a fragment.
func TestClient_FragmentBlocks(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
hldr.SetBit("i", "f", 0, 1)
hldr.SetBit("i", "f", pilosa.HashBlockSize*3, 100)
// Set a bit on a different shard.
hldr.SetBit("i", "f", 0, 1)
c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil))
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", "standard", 0)
if err != nil {
t.Fatal(err)
} else if len(blocks) != 2 {
t.Fatalf("unexpected blocks: %s", spew.Sdump(blocks))
} else if blocks[0].ID != 0 {
t.Fatalf("unexpected block id(0): %d", blocks[0].ID)
} else if blocks[1].ID != 3 {
t.Fatalf("unexpected block id(1): %d", blocks[1].ID)
}
// Verify data matches local blocks.
if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", "standard", 0); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(a, blocks) {
t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks))
}
}
func TestClient_CreateDecimalField(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
cmd := cluster[0]
c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil))
index := "cdf"
err := c.CreateIndex(context.Background(), index, pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
field := "dfield"
err = c.CreateFieldWithOptions(context.Background(), index, field, pilosa.FieldOptions{Type: pilosa.FieldTypeDecimal, Scale: 1, Min: -1000, Max: 1000})
if err != nil {
t.Fatalf("creating field: %v", err)
}
fld, err := cmd.API.Field(context.Background(), index, field)
if err != nil {
t.Fatalf("getting field: %v", err)
}
if fld.Options().Scale != 1 {
t.Fatalf("expected Scale 1, got: %+v", fld.Options())
}
err = c.ImportValue2(context.Background(), &pilosa.ImportValueRequest{Index: index, Field: field, ColumnIDs: []uint64{1, 2, 3}, Shard: 0, FloatValues: []float64{1.1, 2.2, 3.3}}, &pilosa.ImportOptions{})
if err != nil {
t.Fatalf("importing float values: %v", err)
}
resp, err := c.Query(context.Background(), index, &pilosa.QueryRequest{Index: index, Query: "Row(dfield>21)"})
if err != nil {
t.Fatalf("querying: %v", err)
}
if !reflect.DeepEqual(resp.Results[0].(*pilosa.Row).Columns(), []uint64{2, 3}) {
t.Fatalf("unexpected results: %v", resp.Results[0].(*pilosa.Row).Columns())
}
}
// Client represents a test wrapper for pilosa.Client.
type Client struct {
*http.InternalClient
}
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string, h *gohttp.Client) *Client {
c, err := http.NewInternalClient(host, h)
if err != nil {
panic(err)
}
return &Client{InternalClient: c}
}
func makeImportRoaringRequest(clear bool, viewData string) *pilosa.ImportRoaringRequest {
roaringData, _ := hex.DecodeString(viewData)
return &pilosa.ImportRoaringRequest{
Clear: clear,
Views: map[string][]byte{
"": roaringData,
},
}
}
Reference in a new issue View git blame Copy permalink