diff --git a/client_test.go b/client_test.go index aa8b3b126..7466d8639 100644 --- a/client_test.go +++ b/client_test.go @@ -3,13 +3,151 @@ package pilosa_test import ( "bytes" "context" + "fmt" "reflect" "testing" "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/pql" ) +func createCluster(c *pilosa.Cluster) ([]*Server, []*Index) { + numNodes := len(c.Nodes) + idx := make([]*Index, numNodes) + server := make([]*Server, numNodes) + for i := 0; i < numNodes; i++ { + idx[i] = MustOpenIndex() + server[i] = NewServer() + server[i].Handler.Host = server[i].Host() + server[i].Handler.Cluster = c + server[i].Handler.Cluster.Nodes[i].Host = server[i].Host() + server[i].Handler.Index = idx[i].Index + } + return server, idx +} + +// Test distributed TopN Bitmap count across 3 nodes. +func TestClient_MultiNode(t *testing.T) { + cluster := NewCluster(3) + s, idx := createCluster(cluster) + + for i := 0; i < len(cluster.Nodes); i++ { + defer idx[i].Close() + defer s[i].Close() + } + + s[0].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { + e := pilosa.NewExecutor() + e.Index = idx[0].Index + e.Host = cluster.Nodes[0].Host + e.Cluster = cluster + return e.Execute(ctx, db, query, slices, opt) + } + s[1].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { + e := pilosa.NewExecutor() + e.Index = idx[1].Index + e.Host = cluster.Nodes[1].Host + e.Cluster = cluster + return e.Execute(ctx, db, query, slices, opt) + } + s[2].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { + e := pilosa.NewExecutor() + e.Index = idx[2].Index + e.Host = cluster.Nodes[2].Host + e.Cluster = cluster + return e.Execute(ctx, db, query, slices, opt) + } + + // Create a dispersed set of bitmaps across 3 nodes such that each individual node and slice width increment would reveal a different TopN. + idx[0].MustCreateFragmentIfNotExists("d", "f", 0).MustSetBits(99, 1, 2, 3, 4) + idx[0].MustCreateFragmentIfNotExists("d", "f", 0).MustSetBits(100, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10) + idx[0].MustCreateFragmentIfNotExists("d", "f", 0).MustSetBits(98, 1, 2, 3, 4, 5, 6) + idx[0].MustCreateFragmentIfNotExists("d", "f", 0).MustSetBits(1, 4) + idx[0].MustCreateFragmentIfNotExists("d", "f", 0).MustSetBits(22, 1, 2, 3, 4, 5) + + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(100, (SliceWidth*10)+10) + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12) + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12, (SliceWidth*10)+13, (SliceWidth*10)+14, (SliceWidth*10)+15) + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(2, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4) + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(3, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4, (SliceWidth*10)+5) + idx[1].MustCreateFragmentIfNotExists("d", "f", 10).MustSetBits(22, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+10) + + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(24, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13, (SliceWidth*6)+14) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(20, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(21, (SliceWidth*6)+10) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(100, (SliceWidth*6)+10) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(99, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(98, (SliceWidth*6)+10, (SliceWidth*6)+11) + idx[2].MustCreateFragmentIfNotExists("d", "f", 6).MustSetBits(22, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12) + + // Connect to each node to compare results. + client := make([]*Client, 3) + client[0] = MustNewClient(s[0].Host()) + client[1] = MustNewClient(s[0].Host()) + client[2] = MustNewClient(s[0].Host()) + + topN := 4 + q := fmt.Sprintf(`TopN(frame="%s", n=%d)`, "f", topN) + + result, err := client[0].ExecuteQuery(context.Background(), "d", q, true) + if err != nil { + t.Fatal(err) + } + + // Check the results before every node has the correct max slice value. + pairs := result.(internal.QueryResponse).Results[0].Pairs + for _, pair := range pairs { + if pair.Key == 22 && pair.Count != 5 { + t.Fatalf("Invalid Cluster wide MaxSlice prevents accurate calculation of %s", pair) + } + } + + // Set max slice to correct value. + idx[0].DB("d").SetRemoteMaxSlice(10) + idx[1].DB("d").SetRemoteMaxSlice(10) + idx[2].DB("d").SetRemoteMaxSlice(10) + + result, err = client[0].ExecuteQuery(context.Background(), "d", q, true) + if err != nil { + t.Fatal(err) + } + + // Test must return exactly N results. + if len(result.(internal.QueryResponse).Results[0].Pairs) != topN { + t.Fatalf("unexpected number of TopN results: %s", spew.Sdump(result)) + } + p := []*internal.Pair{ + {Key: 100, Count: 12}, + {Key: 22, Count: 11}, + {Key: 98, Count: 8}, + {Key: 99, Count: 7}} + + // Valdidate the Top 4 result counts. + if !reflect.DeepEqual(result.(internal.QueryResponse).Results[0].Pairs, p) { + t.Fatalf("Invalid TopN result set: %s", spew.Sdump(result)) + } + + result1, err := client[1].ExecuteQuery(context.Background(), "d", q, true) + if err != nil { + t.Fatal(err) + } + result2, err := client[2].ExecuteQuery(context.Background(), "d", q, true) + 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 bulk import data. func TestClient_Import(t *testing.T) { idx := MustOpenIndex()