mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
There's a lot going on here. First, we were treating "the test is a Condition" as implying BSI, which it doesn't anymore. Second, the behavior of conditions was weird and BSI-specific. Third, we had to propagate these changes and features throughout a bunch of code, including both the core featurebase code and the DAX replacements/copies of it, plus the SQL3 layer. We refactor this so that tests for equality and inequality work for non-BSI fields, so now if you accidentally use `==` in a Row call on a non-BSI field, it still works; that's not specific to BSI fields anymore. We add a TrackExistence flag to fields, and propagate it through things like our protobuf code, etcetera, so that we can successfully create fields. Newly-created fields get this by default, because we add it unconditionally to them, but the paths that are being called with existing fields don't add it. So, when we "create" (really, just load the definition of) a field from something stored in the schema, we don't add TrackExistence to it, but any path to creating a new field should. A time quantum field with NoStandardView will *effectively* lack TrackExistence. For sets, mutexes, and time quantums with a standard view, anything that sets bits will also set a corresponding bit for the record in a new "existence" view. This allows us to distinguish between an empty set and a null, and also allows null checks to be constant-time. When clearing bits, we don't clear existence bits EXCEPT that if you clear a bit in a mutex, *and the bit actually existed*, we clear the existence bit. For sets and time quantums, clearing bits never clears the existence bit. Deleting records clears the existence bit. We also add code to the `batch` subpackage to generate suitable existence field bitmaps and import them. This logic correctly handles empty sets and nils. The `batch` package does not allow specification of anything equivalent to clearing a single bit from an existing record, so we don't have to deal with the mutex complexity in that case, which is good because it would be impossible. This requires a number of other subtle changes, such as allowing new fields to have more than one FieldOption specified for them. We also drop the handful of implementation bits relating to the "fullySorted" internal-use-only import flag, which existed only to support the JSON ingest API, which we've removed. The most dangerous part of this is that the mutex semantics are impossible to implement on top of our existing API, because they require us to know, not how *many* bits we cleared, but which *specific* bits we cleared. I've implemented this as a new Tx method, which is almost certainly going to be tech debt one day; if we some day drop the Import API, we should remove that. The testing for this is only currently covering the Set/Clear behavior of PQL, and the Import API. The batch tests haven't been written yet. Fields that don't have existence tracking enabled refuse to perform null/not-null tests. They should also report themselves as having no null values -- if a record exists, sets in it are considered empty rather than null. The SQL3 support requires a number of subtle modifications to both featurebase and some addon tooling. The essential thing is dropping the unconditional translation of nil slices to non-nil empty slices in translateResult, both in the executor and the orchestrator. We also modify the logic that handles generating results from Extract calls, to ensure that non-null sets get an empty slice created for them even if they never have any values assigned. The expected results for some tests are different now; we expect to get nil slices, rather than 0-length non-nil slices, for fields which were never written for a given record. Most tests were not changed. (In every case, if a test was failing, I actually checked the logic before changing expected results. This required a lot of tracking down of edge cases.) The batch package now rejects as an error attempts to clear single bits from mutex fields, because so far as I can tell it's simply impossible to have a roaring import that specifies the correct semantics there; you can't tell whether to clear an existence bit without access to the currently-set bits, which the batch API doesn't have. We already supported the special case of specifying a clear value of nil for clearing a mutex field; now that is the only allowed value for a mutex field to have in row.Clears. We change the logic for fixing up incoming view names (in two places) to stop assuming that any view in a time field other than "" that does not have viewStandard as a prefix is a partial time quantum name that should have "standard_" prepended to it. This allows us to submit bitmaps for "existence" to time quantum fields and not have them silently transformed into "standard_existence" because that's what we'd do with "202203". We drop the field ClearBits method, which was totally unused. We drop the sliceDifference function, which was used in a previous mutex implementation and hasn't been used in ages, and the test case for it, and the helper function used only by that test case.
2053 lines
55 KiB
Go
2053 lines
55 KiB
Go
// Copyright 2022 Molecula Corp. (DBA FeatureBase).
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
package server_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
pilosa "github.com/featurebasedb/featurebase/v3"
|
|
"github.com/featurebasedb/featurebase/v3/authn"
|
|
"github.com/featurebasedb/featurebase/v3/authz"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
"github.com/featurebasedb/featurebase/v3/pql"
|
|
pb "github.com/featurebasedb/featurebase/v3/proto"
|
|
"github.com/featurebasedb/featurebase/v3/server"
|
|
"github.com/featurebasedb/featurebase/v3/sql"
|
|
"github.com/featurebasedb/featurebase/v3/test"
|
|
"github.com/featurebasedb/featurebase/v3/vprint"
|
|
"github.com/golang-jwt/jwt"
|
|
"github.com/pkg/errors"
|
|
"github.com/stretchr/testify/assert"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/metadata"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
func TestGRPC(t *testing.T) {
|
|
t.Run("ToTable", func(t *testing.T) {
|
|
type expHeader struct {
|
|
name string
|
|
dataType string
|
|
}
|
|
|
|
type expColumn interface{}
|
|
|
|
va, vb := int64(-11), int64(-12)
|
|
tests := []struct {
|
|
result interface{}
|
|
expHeaders []expHeader
|
|
expColumns [][]expColumn
|
|
}{
|
|
// Row (uint64)
|
|
{
|
|
pilosa.NewRow(10, 11, 12),
|
|
[]expHeader{
|
|
{"_id", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{uint64(10)},
|
|
{uint64(11)},
|
|
{uint64(12)},
|
|
},
|
|
},
|
|
// Row (string)
|
|
{
|
|
&pilosa.Row{Keys: []string{"ten", "eleven", "twelve"}},
|
|
[]expHeader{
|
|
{"_id", "string"},
|
|
},
|
|
[][]expColumn{
|
|
{"ten"},
|
|
{"eleven"},
|
|
{"twelve"},
|
|
},
|
|
},
|
|
// PairField (uint64)
|
|
{
|
|
pilosa.PairField{
|
|
Pair: pilosa.Pair{ID: 10, Count: 123},
|
|
Field: "fld",
|
|
},
|
|
[]expHeader{
|
|
{"fld", "uint64"},
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{uint64(10), uint64(123)},
|
|
},
|
|
},
|
|
// Pair (string)
|
|
{
|
|
pilosa.PairField{
|
|
Pair: pilosa.Pair{Key: "ten", Count: 123},
|
|
Field: "fld",
|
|
},
|
|
[]expHeader{
|
|
{"fld", "string"},
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{string("ten"), uint64(123)},
|
|
},
|
|
},
|
|
// *PairsField (uint64)
|
|
{
|
|
&pilosa.PairsField{
|
|
Pairs: []pilosa.Pair{
|
|
{ID: 10, Count: 123},
|
|
{ID: 11, Count: 456},
|
|
},
|
|
Field: "fld",
|
|
},
|
|
[]expHeader{
|
|
{"fld", "uint64"},
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{uint64(10), uint64(123)},
|
|
{uint64(11), uint64(456)},
|
|
},
|
|
},
|
|
// *PairsField (string)
|
|
{
|
|
&pilosa.PairsField{
|
|
Pairs: []pilosa.Pair{
|
|
{Key: "ten", Count: 123},
|
|
{Key: "eleven", Count: 456},
|
|
},
|
|
Field: "fld",
|
|
},
|
|
[]expHeader{
|
|
{"fld", "string"},
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{"ten", uint64(123)},
|
|
{"eleven", uint64(456)},
|
|
},
|
|
},
|
|
// []GroupCount (uint64)
|
|
{
|
|
pilosa.NewGroupCounts("", []pilosa.GroupCount{
|
|
{
|
|
Group: []pilosa.FieldRow{
|
|
{Field: "a", RowID: 10},
|
|
{Field: "b", RowID: 11},
|
|
},
|
|
Count: 123,
|
|
},
|
|
{
|
|
Group: []pilosa.FieldRow{
|
|
{Field: "a", RowID: 10},
|
|
{Field: "b", RowID: 12},
|
|
},
|
|
Count: 456,
|
|
},
|
|
{
|
|
Group: []pilosa.FieldRow{
|
|
{Field: "va", Value: &va},
|
|
{Field: "vb", Value: &vb},
|
|
},
|
|
Count: 789,
|
|
},
|
|
}...),
|
|
[]expHeader{
|
|
{"a", "uint64"},
|
|
{"b", "uint64"},
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{uint64(10), uint64(11), uint64(123)},
|
|
{uint64(10), uint64(12), uint64(456)},
|
|
{int64(va), int64(vb), uint64(789)},
|
|
},
|
|
},
|
|
// []GroupCount (string) + sum
|
|
{
|
|
pilosa.NewGroupCounts("sum", []pilosa.GroupCount{
|
|
{
|
|
Group: []pilosa.FieldRow{
|
|
{Field: "a", RowKey: "ten"},
|
|
{Field: "b", RowKey: "eleven"},
|
|
},
|
|
Count: 123,
|
|
},
|
|
{
|
|
Group: []pilosa.FieldRow{
|
|
{Field: "a", RowKey: "ten"},
|
|
{Field: "b", RowKey: "twelve"},
|
|
},
|
|
Count: 456,
|
|
},
|
|
}...),
|
|
[]expHeader{
|
|
{"a", "string"},
|
|
{"b", "string"},
|
|
{"count", "uint64"},
|
|
{"sum", "int64"},
|
|
},
|
|
[][]expColumn{
|
|
{"ten", "eleven", uint64(123), int64(0)},
|
|
{"ten", "twelve", uint64(456), int64(0)},
|
|
},
|
|
},
|
|
// RowIdentifiers (uint64)
|
|
{
|
|
pilosa.RowIdentifiers{
|
|
Rows: []uint64{10, 11, 12},
|
|
},
|
|
[]expHeader{
|
|
{"", "uint64"}, // This is blank because we don't expose RowIdentifiers.field, so we have no way to set it for tests.
|
|
},
|
|
[][]expColumn{
|
|
{uint64(10)},
|
|
{uint64(11)},
|
|
{uint64(12)},
|
|
},
|
|
},
|
|
// RowIdentifiers (string)
|
|
{
|
|
pilosa.RowIdentifiers{
|
|
Keys: []string{"ten", "eleven", "twelve"},
|
|
},
|
|
[]expHeader{
|
|
{"", "string"}, // This is blank because we don't expose RowIdentifiers.field, so we have no way to set it for tests.
|
|
},
|
|
[][]expColumn{
|
|
{"ten"},
|
|
{"eleven"},
|
|
{"twelve"},
|
|
},
|
|
},
|
|
// uint64
|
|
{
|
|
uint64(123),
|
|
[]expHeader{
|
|
{"count", "uint64"},
|
|
},
|
|
[][]expColumn{
|
|
{uint64(123)},
|
|
},
|
|
},
|
|
// bool
|
|
{
|
|
true,
|
|
[]expHeader{
|
|
{"result", "bool"},
|
|
},
|
|
[][]expColumn{
|
|
{true},
|
|
},
|
|
},
|
|
// ValCount
|
|
{
|
|
pilosa.ValCount{Val: 1, Count: 1},
|
|
[]expHeader{{"value", "int64"}, {"count", "int64"}},
|
|
[][]expColumn{{int64(1), int64(1)}},
|
|
},
|
|
{
|
|
pilosa.ValCount{FloatVal: 1.24, Count: 1},
|
|
[]expHeader{{"value", "float64"}, {"count", "int64"}},
|
|
[][]expColumn{{float64(1.24), int64(1)}},
|
|
},
|
|
// SignedRow
|
|
{
|
|
pilosa.SignedRow{
|
|
Neg: pilosa.NewRow(13, 14, 15),
|
|
Pos: pilosa.NewRow(10, 11, 12),
|
|
},
|
|
[]expHeader{
|
|
{"", "int64"},
|
|
},
|
|
[][]expColumn{
|
|
{int64(-15)},
|
|
{int64(-14)},
|
|
{int64(-13)},
|
|
{int64(10)},
|
|
{int64(11)},
|
|
{int64(12)},
|
|
},
|
|
},
|
|
}
|
|
|
|
for ti, test := range tests {
|
|
toTabler, err := server.ToTablerWrapper(test.result)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
table, err := toTabler.ToTable()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Ensure headers match.
|
|
headers := table.GetHeaders()
|
|
if len(headers) < len(test.expHeaders) {
|
|
t.Fatalf("test %d expected %d headers, got %d, first missing header %q",
|
|
ti, len(test.expHeaders), len(headers), test.expHeaders[len(headers)].name)
|
|
}
|
|
for i, header := range headers {
|
|
if header.Name != test.expHeaders[i].name {
|
|
t.Fatalf("test %d expected header name: %s, but got: %s", ti, test.expHeaders[i].name, header.Name)
|
|
}
|
|
if header.Datatype != test.expHeaders[i].dataType {
|
|
t.Fatalf("test %d expected header data type: %s, but got: %s", ti, test.expHeaders[i].dataType, header.Datatype)
|
|
}
|
|
}
|
|
|
|
// Ensure column data matches.
|
|
for i, row := range table.GetRows() {
|
|
columns := row.GetColumns()
|
|
if len(columns) != len(test.expColumns[i]) {
|
|
t.Fatalf("test %d expected %d columns, got %d in row %d",
|
|
ti, len(test.expColumns[i]), len(columns), i)
|
|
}
|
|
for j, column := range columns {
|
|
switch v := test.expColumns[i][j].(type) {
|
|
case string:
|
|
val := column.GetStringVal()
|
|
if val != v {
|
|
t.Fatalf("test %d expected column val: %v, but got: %v", ti, v, val)
|
|
}
|
|
case uint64:
|
|
val := column.GetUint64Val()
|
|
if val != v {
|
|
t.Fatalf("test %d expected column val: %v, but got: %v", ti, v, val)
|
|
}
|
|
case bool:
|
|
val := column.GetBoolVal()
|
|
if val != v {
|
|
t.Fatalf("test %d expected column val: %v, but got: %v", ti, v, val)
|
|
}
|
|
case int64:
|
|
val := column.GetInt64Val()
|
|
if val != v {
|
|
t.Fatalf("test %d expected column val: %v but got: %v", ti, v, val)
|
|
}
|
|
case float64:
|
|
val := column.GetFloat64Val()
|
|
if val != v {
|
|
t.Fatalf("test %d expected column val: %v but got: %v", ti, v, val)
|
|
}
|
|
default:
|
|
t.Fatalf("test %d has unhandled data type: %T", ti, v)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestQueryPQLUnary(t *testing.T) {
|
|
m := test.RunCommand(t)
|
|
defer m.Close()
|
|
|
|
i := m.MustCreateIndex(t, "i", pilosa.IndexOptions{})
|
|
m.MustCreateField(t, i.Name(), "f", pilosa.OptFieldKeys())
|
|
gh := server.NewGRPCHandler(m.API)
|
|
|
|
stream := &MockServerTransportStream{}
|
|
ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
|
|
|
|
resp, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: i.Name(),
|
|
Pql: `Set(0, f="zero")`,
|
|
})
|
|
if err != nil {
|
|
// Unary query should work
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if resp.Duration == 0 {
|
|
t.Fatal("duration not recorded")
|
|
}
|
|
duration, err := stream.GetDuration()
|
|
if duration == 0 || err != nil {
|
|
t.Fatal("duration header not recorded")
|
|
}
|
|
|
|
_, err = gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: i.Name(),
|
|
Pql: `Set(1, f="one") Set(2, f="two")`,
|
|
})
|
|
staterr := status.Convert(err)
|
|
if staterr == nil || staterr.Code() != codes.InvalidArgument {
|
|
// QueryPQLUnary handles exactly one query
|
|
t.Fatalf("expected error: InvalidArgument, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestQueryPQL(t *testing.T) {
|
|
// TODO: Replace TestQueryPQL and TestQueryPQLUnary with table-driven test, like TestQuerySQL
|
|
m := test.RunCommand(t)
|
|
defer m.Close()
|
|
|
|
i := m.MustCreateIndex(t, "i", pilosa.IndexOptions{Keys: false, TrackExistence: true})
|
|
m.MustCreateField(t, i.Name(), "f", pilosa.OptFieldKeys())
|
|
gh := server.NewGRPCHandler(m.API)
|
|
|
|
mock := &mockPilosa_QuerySQLServer{ctx: context.Background()}
|
|
|
|
err := gh.QueryPQL(&pb.QueryPQLRequest{
|
|
Index: i.Name(),
|
|
Pql: `Set(0, f="zero")`,
|
|
}, mock)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
duration, err := mock.GetDuration()
|
|
if duration == 0 || err != nil {
|
|
t.Fatal("duration header not recorded")
|
|
}
|
|
|
|
if len(mock.Results) != 1 {
|
|
t.Fatal("expecting one result")
|
|
}
|
|
|
|
if len(mock.Results[0].Headers) != 1 {
|
|
t.Fatal("expecting one header")
|
|
}
|
|
|
|
if len(mock.Results[0].Columns) != 1 {
|
|
t.Fatal("expecting one column")
|
|
}
|
|
|
|
if mock.Results[0].Duration == 0 {
|
|
t.Fatal("expecting non-zero duration")
|
|
}
|
|
|
|
// Set second value so that All() returns more than one result
|
|
err = gh.QueryPQL(&pb.QueryPQLRequest{
|
|
Index: i.Name(),
|
|
Pql: `Set(1, f="zero")`,
|
|
}, mock)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
mock.clearResults()
|
|
|
|
err = gh.QueryPQL(&pb.QueryPQLRequest{
|
|
Index: i.Name(),
|
|
Pql: `All()`,
|
|
}, mock)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
if len(mock.Results) != 2 {
|
|
t.Fatal("expecting two results")
|
|
}
|
|
|
|
if len(mock.Results[0].Headers) != 1 {
|
|
t.Fatal("expecting one header")
|
|
}
|
|
|
|
if len(mock.Results[0].Columns) != 1 {
|
|
t.Fatal("expecting one column")
|
|
}
|
|
|
|
if len(mock.Results[1].Headers) != 0 {
|
|
t.Fatal("expecting no headers on second result")
|
|
}
|
|
|
|
if len(mock.Results[1].Columns) != 1 {
|
|
t.Fatal("expecting one column on second result")
|
|
}
|
|
|
|
if mock.Results[0].Duration == 0 {
|
|
t.Fatal("expecting non-zero duration")
|
|
}
|
|
|
|
if mock.Results[1].Duration != 0 {
|
|
t.Fatal("expecting zero duration on second result")
|
|
}
|
|
}
|
|
|
|
type (
|
|
tableResponse struct {
|
|
headers []columnInfo
|
|
rows []row
|
|
}
|
|
columnInfo struct {
|
|
name string
|
|
datatype string
|
|
}
|
|
row struct {
|
|
columns []columnResponse
|
|
}
|
|
columnResponse interface{}
|
|
)
|
|
|
|
func TestQuerySQL(t *testing.T) {
|
|
|
|
stream := &MockServerTransportStream{}
|
|
ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
|
|
|
|
gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t)
|
|
defer tearDownFunc()
|
|
|
|
tests := []struct {
|
|
sql string
|
|
exp tableResponse
|
|
eq func(tableResponse, tableResponse) error
|
|
wantErr string
|
|
}{
|
|
{
|
|
// Extract(Limit(All(), limit=100, offset=0),Rows(age))
|
|
sql: "select age from grouper",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(27)}},
|
|
{[]columnResponse{int64(16)}},
|
|
{[]columnResponse{int64(19)}},
|
|
{[]columnResponse{int64(27)}},
|
|
{[]columnResponse{int64(16)}},
|
|
{[]columnResponse{int64(34)}},
|
|
{[]columnResponse{int64(27)}},
|
|
{[]columnResponse{int64(16)}},
|
|
{[]columnResponse{int64(16)}},
|
|
{[]columnResponse{int64(31)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// Extract(Limit(ConstRow(columns=[2]), limit=100, offset=0),Rows(age),Rows(color),Rows(height),Rows(score))
|
|
sql: "select * from grouper where _id=2",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"_id", "uint64"},
|
|
{"age", "int64"},
|
|
{"color", "[]string"},
|
|
{"height", "int64"},
|
|
{"score", "int64"},
|
|
{"timestamp", "timestamp"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(2), int64(16), []string{"blue"}, int64(30), int64(-8), "2011-01-02T12:32:00Z"}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// Extract(Limit(ConstRow(columns=[2]), limit=100, offset=0),Rows(age),Rows(color),Rows(height),Rows(score))
|
|
sql: "select * from grouper",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"_id", "uint64"},
|
|
{"age", "int64"},
|
|
{"color", "[]string"},
|
|
{"height", "int64"},
|
|
{"score", "int64"},
|
|
{"timestamp", "timestamp"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(1), int64(27), []string{"blue"}, int64(20), int64(-10), "2011-04-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(2), int64(16), []string{"blue"}, int64(30), int64(-8), "2011-01-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(3), int64(19), []string{"red"}, int64(40), int64(6), "2012-01-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(4), int64(27), []string{"green"}, int64(50), int64(0), "2013-09-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(5), int64(16), []string{"blue"}, int64(60), int64(-2), "2014-01-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(6), int64(34), []string{"blue"}, int64(70), int64(100), "2010-05-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(7), int64(27), []string{"blue"}, int64(80), int64(0), "2016-08-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(8), int64(16), []string(nil), int64(90), int64(-13), "2020-01-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(9), int64(16), []string{"red"}, int64(100), int64(80), "2000-03-02T12:32:00Z"}},
|
|
{[]columnResponse{uint64(10), int64(31), []string{"red"}, int64(110), int64(-2), "2018-01-02T12:32:00Z"}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
// join
|
|
{
|
|
// Count(Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')))
|
|
sql: "select count(*) from grouper g INNER JOIN joiner j ON g._id = j.grouperid",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"count(*)", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(8)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid'))
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(1)}},
|
|
{[]columnResponse{uint64(2)}},
|
|
{[]columnResponse{uint64(3)}},
|
|
{[]columnResponse{uint64(5)}},
|
|
{[]columnResponse{uint64(6)}},
|
|
{[]columnResponse{uint64(7)}},
|
|
{[]columnResponse{uint64(8)}},
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// Intersect(Row(color='red'),Distinct(Row(grouperid!=null),index='joiner',field='grouperid'))
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red'",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(3)}},
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// Intersect(Row(color='red'),Distinct(Row(jointype=2),index='joiner',field='grouperid'))
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and j.jointype = 2",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(3)}},
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
// order by
|
|
{
|
|
// Distinct(Row(score!=null),index='grouper',field='score')
|
|
sql: "select distinct score from grouper order by score asc",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"score", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(-13)}},
|
|
{[]columnResponse{int64(-10)}},
|
|
{[]columnResponse{int64(-8)}},
|
|
{[]columnResponse{int64(-2)}},
|
|
{[]columnResponse{int64(0)}},
|
|
{[]columnResponse{int64(6)}},
|
|
{[]columnResponse{int64(80)}},
|
|
{[]columnResponse{int64(100)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// Distinct(Row(score!=null),index='grouper',field='score')
|
|
sql: "select distinct score from grouper order by score desc",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"score", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(100)}},
|
|
{[]columnResponse{int64(80)}},
|
|
{[]columnResponse{int64(6)}},
|
|
{[]columnResponse{int64(0)}},
|
|
{[]columnResponse{int64(-2)}},
|
|
{[]columnResponse{int64(-8)}},
|
|
{[]columnResponse{int64(-10)}},
|
|
{[]columnResponse{int64(-13)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// Distinct(Row(score!=null),index='grouper',field='score')
|
|
sql: "select distinct score from grouper order by score asc limit 5",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"score", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(-13)}},
|
|
{[]columnResponse{int64(-10)}},
|
|
{[]columnResponse{int64(-8)}},
|
|
{[]columnResponse{int64(-2)}},
|
|
{[]columnResponse{int64(0)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
|
|
{
|
|
// Distinct(Row(score!=null),index='grouper',field='score')
|
|
sql: "select distinct score from grouper order by score desc limit 5",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"score", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(100)}},
|
|
{[]columnResponse{int64(80)}},
|
|
{[]columnResponse{int64(6)}},
|
|
{[]columnResponse{int64(0)}},
|
|
{[]columnResponse{int64(-2)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
|
|
// distinct
|
|
{
|
|
// Distinct(Row(score!=null),index='grouper',field='score')
|
|
sql: "select distinct score from grouper",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"score", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(-13)}},
|
|
{[]columnResponse{int64(-10)}},
|
|
{[]columnResponse{int64(-8)}},
|
|
{[]columnResponse{int64(-2)}},
|
|
{[]columnResponse{int64(0)}},
|
|
{[]columnResponse{int64(6)}},
|
|
{[]columnResponse{int64(80)}},
|
|
{[]columnResponse{int64(100)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
|
|
// Distinct(Row(height!=null),index='grouper',field='height')
|
|
sql: "select distinct height from grouper",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"height", "int64"}},
|
|
rows: []row{
|
|
{[]columnResponse{int64(20)}},
|
|
{[]columnResponse{int64(30)}},
|
|
{[]columnResponse{int64(40)}},
|
|
{[]columnResponse{int64(50)}},
|
|
{[]columnResponse{int64(60)}},
|
|
{[]columnResponse{int64(70)}},
|
|
{[]columnResponse{int64(80)}},
|
|
{[]columnResponse{int64(90)}},
|
|
{[]columnResponse{int64(100)}},
|
|
{[]columnResponse{int64(110)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
|
|
// groupby
|
|
{
|
|
// GroupBy(Rows(field='age'),limit=100)
|
|
sql: "select age as yrs, count(*) as cnt from grouper group by age",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"yrs", "int64"},
|
|
{"cnt", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), uint64(4)}},
|
|
{[]columnResponse{int64(19), uint64(1)}},
|
|
{[]columnResponse{int64(27), uint64(3)}},
|
|
{[]columnResponse{int64(31), uint64(1)}},
|
|
{[]columnResponse{int64(34), uint64(1)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),Rows(field='color'),limit=100)
|
|
sql: "select age, color, count(*) from grouper group by age, color",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"color", "string"},
|
|
{"count(*)", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), "blue", uint64(2)}},
|
|
{[]columnResponse{int64(16), "red", uint64(1)}},
|
|
{[]columnResponse{int64(19), "red", uint64(1)}},
|
|
{[]columnResponse{int64(27), "blue", uint64(2)}},
|
|
{[]columnResponse{int64(27), "green", uint64(1)}},
|
|
{[]columnResponse{int64(31), "red", uint64(1)}},
|
|
{[]columnResponse{int64(34), "blue", uint64(1)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),Rows(field='color'),limit=100,filter=Row(age=27),aggregate=Sum(field='height'))
|
|
sql: "select age, color, sum(height) from grouper where age = 27 group by age, color",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"color", "string"},
|
|
{"sum(height)", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(27), "blue", int64(100)}},
|
|
{[]columnResponse{int64(27), "green", int64(50)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),limit=100,having=Condition(count>1))
|
|
sql: "select age, count(*) from grouper group by age having count > 1",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"count(*)", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), uint64(4)}},
|
|
{[]columnResponse{int64(27), uint64(3)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),limit=100,having=Condition(1<=count<=3))
|
|
sql: "select age, count(*) from grouper group by age having count between 1 and 3",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"count(*)", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(19), uint64(1)}},
|
|
{[]columnResponse{int64(27), uint64(3)}},
|
|
{[]columnResponse{int64(31), uint64(1)}},
|
|
{[]columnResponse{int64(34), uint64(1)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),limit=3)
|
|
sql: "select age, count(*) as cnt from grouper group by age order by cnt desc, age desc limit 3",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"cnt", "uint64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), uint64(4)}},
|
|
{[]columnResponse{int64(27), uint64(3)}},
|
|
{[]columnResponse{int64(19), uint64(1)}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
// GroupBy(Rows(field='age'),Rows(field='height'),filter=Intersect(Row(timestamp>"2017-09-02T12:32:00Z"),Row(height>40)))
|
|
sql: "select age, height from grouper where timestamp > '2017-09-02T12:32:00Z' and height > 40 group by age, height",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"height", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), int64(90)}},
|
|
{[]columnResponse{int64(31), int64(110)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// Extract(Union(Row(timestamp>"2017-09-02T12:32:00Z"),Row(height>90)),Rows(age), Rows(height))
|
|
sql: "select age, height from grouper where timestamp > '2017-09-02T12:32:00Z' or height > 90",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"height", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16), int64(90)}},
|
|
{[]columnResponse{int64(16), int64(100)}},
|
|
{[]columnResponse{int64(31), int64(110)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
//Extract(Intersect(Row(timestamp>"2017-09-02T12:32:00Z"),Row(timestamp<"2019-09-02T12:32:00Z")),Rows(age), Rows(height))
|
|
sql: "select age, height from grouper where timestamp > '2017-09-02T12:32:00Z' and timestamp < '2019-09-02T12:32:00Z'",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"height", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(31), int64(110)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
//Extract(Intersect(Row(timestamp>"2017-09-02T12:32:00Z"),Row(timestamp<"2019-09-02T12:32:00Z")),Rows(age), Rows(height))
|
|
//Testing the parenthesis around where clause
|
|
sql: "select age, height from grouper where (timestamp > '2017-09-02T12:32:00Z' and timestamp < '2019-09-02T12:32:00Z')",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
{"height", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(31), int64(110)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
//Testing empty parenthesis around where clause
|
|
sql: "select age, height from grouper where ()",
|
|
wantErr: "parsing sql",
|
|
},
|
|
{
|
|
//Distinct(Row(timestamp>"2019-09-02T12:32:00Z"), index='grouper',field='age')
|
|
sql: "select distinct age from grouper where timestamp > '2019-09-02T12:32:00Z'",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"age", "int64"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{int64(16)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
sql: "show tables",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"Table", "string"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{"another_one"}},
|
|
{[]columnResponse{"deletable_index"}},
|
|
{[]columnResponse{"delete_me"}},
|
|
{[]columnResponse{"grouper"}},
|
|
{[]columnResponse{"joiner"}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
sql: "show fields from grouper",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"Field", "string"},
|
|
{"Type", "string"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{"age", "int"}},
|
|
{[]columnResponse{"color", "keyed-set"}},
|
|
{[]columnResponse{"height", "int"}},
|
|
{[]columnResponse{"score", "int"}},
|
|
{[]columnResponse{"timestamp", "timestamp"}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
sql: "drop table delete_me",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{},
|
|
rows: []row{},
|
|
},
|
|
eq: equal,
|
|
},
|
|
{
|
|
sql: "show tables",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{
|
|
{"Table", "string"},
|
|
},
|
|
rows: []row{
|
|
{[]columnResponse{"another_one"}},
|
|
{[]columnResponse{"deletable_index"}},
|
|
|
|
{[]columnResponse{"grouper"}},
|
|
{[]columnResponse{"joiner"}},
|
|
},
|
|
},
|
|
eq: equal,
|
|
},
|
|
// The following cases test different paths within the `case *sqlparser.AndExpr`
|
|
// of extract.go by providing different WHERE conditions.
|
|
{
|
|
// len(left) == 2 && len(right) == 1
|
|
// right[0].table == left[0].table
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and j.jointype = 2 and g.age = 16",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// len(left) == 2 && len(right) == 1
|
|
// right[0].table == left[1].table {
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where j.jointype = 2 and g.color = 'red' and g.age = 16",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
{
|
|
// len(left) == 1 && len(right) == 1 && left[0].table != right[0].table
|
|
sql: "select _id from grouper g INNER JOIN joiner j ON g._id = j.grouperid where g.color = 'red' and g.age = 16 and j.jointype = 2",
|
|
exp: tableResponse{
|
|
headers: []columnInfo{{"_id", "uint64"}},
|
|
rows: []row{
|
|
{[]columnResponse{uint64(9)}},
|
|
},
|
|
},
|
|
eq: equalUnordered,
|
|
},
|
|
}
|
|
|
|
for i, test := range tests {
|
|
t.Run("test-"+strconv.Itoa(i), func(t *testing.T) {
|
|
resp, err := gh.QuerySQLUnary(ctx, &pb.QuerySQLRequest{Sql: test.sql})
|
|
if test.wantErr != "" {
|
|
if err == nil {
|
|
t.Errorf("expected error %q, got nil", test.wantErr)
|
|
} else if !strings.Contains(err.Error(), test.wantErr) {
|
|
t.Errorf("expected error %q, got %q", test.wantErr, err)
|
|
}
|
|
return
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("sql: %s, error: %v", test.sql, err)
|
|
} else {
|
|
if resp.Duration == 0 {
|
|
t.Fatal("duration not recorded")
|
|
}
|
|
duration, err := stream.GetDuration()
|
|
if duration == 0 || err != nil {
|
|
t.Fatal("duration header not recorded")
|
|
}
|
|
stream.ClearMD()
|
|
tr := toTableResponse(resp)
|
|
if err := test.eq(test.exp, tr); err != nil {
|
|
t.Fatalf("sql: %s, error: %+v", test.sql, err)
|
|
}
|
|
}
|
|
})
|
|
t.Run("test-"+strconv.Itoa(i)+"-streaming", func(t *testing.T) {
|
|
if strings.HasPrefix(test.sql, "drop table") {
|
|
t.Skip("drop statements can only run once")
|
|
}
|
|
mock := &mockPilosa_QuerySQLServer{ctx: context.Background()}
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: test.sql}, mock)
|
|
if test.wantErr != "" {
|
|
if err == nil {
|
|
t.Errorf("expected error %q, got nil", test.wantErr)
|
|
} else if !strings.Contains(err.Error(), test.wantErr) {
|
|
t.Errorf("expected error %q, got %q", test.wantErr, err)
|
|
}
|
|
return
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("sql: %s, error: %v", test.sql, err)
|
|
} else {
|
|
if mock.Results[0].Duration == 0 {
|
|
t.Fatal("duration not recorded")
|
|
}
|
|
duration, err := mock.GetDuration()
|
|
if duration == 0 || err != nil {
|
|
t.Fatal("duration header not recorded")
|
|
}
|
|
if len(mock.Results) > 1 && mock.Results[1].Duration != 0 {
|
|
t.Fatal("duration on second result expected to be zero")
|
|
}
|
|
// TODO: test result values
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestQuerySQLWithError(t *testing.T) {
|
|
|
|
stream := &MockServerTransportStream{}
|
|
ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
|
|
gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t)
|
|
defer tearDownFunc()
|
|
tests := []struct {
|
|
sql string
|
|
err error
|
|
}{
|
|
{
|
|
|
|
sql: "select * from index_not_found",
|
|
err: pilosa.ErrIndexNotFound,
|
|
},
|
|
{
|
|
|
|
sql: "select field_not_found from grouper",
|
|
err: pilosa.ErrFieldNotFound,
|
|
},
|
|
{
|
|
|
|
sql: "select * from grouper, index_not_found",
|
|
err: sql.ErrUnsupportedQuery,
|
|
},
|
|
{
|
|
|
|
sql: "select _id, age, field_not_found from grouper",
|
|
err: pilosa.ErrFieldNotFound,
|
|
},
|
|
{
|
|
sql: "select age, color, count(*) from grouper group by field_not_found, age, color",
|
|
err: pilosa.ErrFieldNotFound,
|
|
},
|
|
{
|
|
sql: "select count(*) from grouper inner join joiner on grouper._id = joiner.field_not_found",
|
|
err: pilosa.ErrFieldNotFound,
|
|
},
|
|
}
|
|
|
|
for i, test := range tests {
|
|
t.Run("test-"+strconv.Itoa(i), func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(ctx, &pb.QuerySQLRequest{Sql: test.sql})
|
|
if err == nil {
|
|
t.Fatalf("sql: %s, expected error: %v", test.sql, test.err)
|
|
} else if errors.Cause(err) != test.err {
|
|
t.Fatalf("sql: %s, expected error: %v, got: %v", test.sql, test.err, err)
|
|
}
|
|
})
|
|
}
|
|
permissions := `
|
|
"user-groups":
|
|
"dca35310-ecda-4f23-86cd-876aee55906b":
|
|
"grouper": "read"
|
|
"dca35310-ecda-4f23-86cd-876aee55906f":
|
|
"grouper": "write"
|
|
admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
|
|
|
|
permFile := writeTestFile(t, "permissions.yaml", permissions)
|
|
auth := server.Auth{
|
|
Enable: true,
|
|
ClientId: "e9088663-eb08-41d7-8f65-efb5f54bbb71",
|
|
ClientSecret: "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF",
|
|
AuthorizeURL: "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/authorize",
|
|
TokenURL: "https://login.microsoftonline.com/4a137d66-d161-4ae4-b1e6-07e9920874b8/oauth2/v2.0/token",
|
|
GroupEndpointURL: "https://graph.microsoft.com/v1.0/me/transitiveMemberOf/microsoft.graph.group?$count=true",
|
|
LogoutURL: "https://login.microsoftonline.com/common/oauth2/v2.0/logout",
|
|
Scopes: []string{"https://graph.microsoft.com/.default", "offline_access"},
|
|
SecretKey: "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF",
|
|
PermissionsFile: permFile,
|
|
}
|
|
var p authz.GroupPermissions
|
|
permsFile, err := os.Open(permFile)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer permsFile.Close()
|
|
|
|
if err = p.ReadPermissionsFile(permsFile); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
gh = gh.WithPerms(&p)
|
|
makeUser := func(groups []authn.Group, name string) *authn.UserInfo {
|
|
// make a valid token
|
|
tkn := jwt.New(jwt.SigningMethodHS256)
|
|
claims := tkn.Claims.(jwt.MapClaims)
|
|
claims["oid"] = "42"
|
|
claims["name"] = name
|
|
secretKey, _ := hex.DecodeString(auth.SecretKey)
|
|
|
|
validToken, err := tkn.SignedString(secretKey)
|
|
if err != nil {
|
|
t.Fatalf("unexpected error creating token %v", err)
|
|
}
|
|
validToken = "Bearer " + validToken
|
|
|
|
return &authn.UserInfo{
|
|
UserID: "fake" + name,
|
|
UserName: name,
|
|
Groups: groups,
|
|
Token: validToken,
|
|
Expiry: time.Time{},
|
|
}
|
|
}
|
|
|
|
user := makeUser([]authn.Group{{GroupID: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", GroupName: "adminGroup"}}, "admin")
|
|
adminCtx := authn.WithUserInfo(ctx, user)
|
|
readuser := makeUser([]authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906b", GroupName: "readers"}}, "reader")
|
|
readCtx := authn.WithUserInfo(ctx, readuser)
|
|
writeuser := makeUser([]authn.Group{{GroupID: "dca35310-ecda-4f23-86cd-876aee55906f", GroupName: "writers"}}, "admin")
|
|
writeCtx := authn.WithUserInfo(ctx, writeuser)
|
|
|
|
sql := "select * from grouper"
|
|
t.Run("test-auth-with-admin-sqlUnary", func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(adminCtx, &pb.QuerySQLRequest{Sql: sql})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-auth-with-read-sqlUnary", func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(readCtx, &pb.QuerySQLRequest{Sql: sql})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-admin-auth-sql", func(t *testing.T) {
|
|
mock := &mockPilosa_QuerySQLServer{ctx: adminCtx}
|
|
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: sql}, mock)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("test-admin-auth-sql-show", func(t *testing.T) {
|
|
mock := &mockPilosa_QuerySQLServer{ctx: adminCtx}
|
|
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: "show tables"}, mock)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("test-admin-auth-get-index", func(t *testing.T) {
|
|
_, err := gh.GetIndex(adminCtx, &pb.GetIndexRequest{Name: "grouper"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-admin-auth-get-indexes", func(t *testing.T) {
|
|
|
|
_, err := gh.GetIndexes(adminCtx, &pb.GetIndexesRequest{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-admin-auth-pql", func(t *testing.T) {
|
|
_, err := gh.QueryPQLUnary(adminCtx, &pb.QueryPQLRequest{
|
|
Index: "grouper",
|
|
Pql: `Set(0, color="red")`,
|
|
})
|
|
if err != nil {
|
|
// Unary query should work
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-write-with-read-auth-pql", func(t *testing.T) {
|
|
_, err := gh.QueryPQLUnary(readCtx, &pb.QueryPQLRequest{
|
|
Index: "grouper",
|
|
Pql: `Set(0, color="red")`,
|
|
})
|
|
if err == nil {
|
|
//should not be able to write
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-show-tables-unary", func(t *testing.T) {
|
|
response, err := gh.QuerySQLUnary(readCtx, &pb.QuerySQLRequest{
|
|
Sql: "show tables",
|
|
})
|
|
|
|
if err != nil && len(response.Rows) != 1 {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("test-show-tables-unary-admin", func(t *testing.T) {
|
|
response, err := gh.QuerySQLUnary(adminCtx, &pb.QuerySQLRequest{
|
|
Sql: "show tables",
|
|
})
|
|
if err != nil && len(response.Rows) != 3 {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-write-with-write-auth-pql", func(t *testing.T) {
|
|
_, err := gh.QueryPQLUnary(writeCtx, &pb.QueryPQLRequest{
|
|
Index: "grouper",
|
|
Pql: `Set(0, color="red")`,
|
|
})
|
|
if err != nil {
|
|
//should be able to write
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
t.Run("test-write-with-admin-auth-pql", func(t *testing.T) {
|
|
_, err := gh.QueryPQLUnary(adminCtx, &pb.QueryPQLRequest{
|
|
Index: "grouper",
|
|
Pql: `Set(0, color="green")`,
|
|
})
|
|
if err != nil {
|
|
//should be able to write
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("test-drop-table-unary-read", func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(readCtx, &pb.QuerySQLRequest{
|
|
Sql: "drop table deletable_index",
|
|
})
|
|
if err == nil {
|
|
t.Fatal("expected error but got nil")
|
|
}
|
|
})
|
|
|
|
t.Run("test-drop-table-unary-write", func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(writeCtx, &pb.QuerySQLRequest{
|
|
Sql: "drop table deletable_index",
|
|
})
|
|
if err == nil {
|
|
t.Fatal("expected error but got nil")
|
|
}
|
|
})
|
|
|
|
t.Run("test-drop-table-unary-admin", func(t *testing.T) {
|
|
_, err := gh.QuerySQLUnary(adminCtx, &pb.QuerySQLRequest{
|
|
Sql: "drop table deletable_index",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("expected nil error but got %v", err)
|
|
}
|
|
})
|
|
|
|
t.Run("test-drop-table-stream-read", func(t *testing.T) {
|
|
mock := &mockPilosa_QuerySQLServer{ctx: readCtx}
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: "drop table another_one"}, mock)
|
|
if err == nil {
|
|
t.Fatal("expected error but got nil")
|
|
}
|
|
})
|
|
t.Run("test-drop-table-stream-write", func(t *testing.T) {
|
|
mock := &mockPilosa_QuerySQLServer{ctx: writeCtx}
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: "drop table another_one"}, mock)
|
|
if err == nil {
|
|
t.Fatal("expected error but got nil")
|
|
}
|
|
})
|
|
t.Run("test-drop-table-stream-admin", func(t *testing.T) {
|
|
mock := &mockPilosa_QuerySQLServer{ctx: adminCtx}
|
|
err := gh.QuerySQL(&pb.QuerySQLRequest{Sql: "drop table another_one"}, mock)
|
|
if err != nil {
|
|
t.Fatalf("expected nil error but got %v", err)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestCRUDIndexes(t *testing.T) {
|
|
m := test.RunCommand(t)
|
|
defer m.Close()
|
|
|
|
stream := &MockServerTransportStream{}
|
|
ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream)
|
|
gh := server.NewGRPCHandler(m.API)
|
|
t.Run("CreateIndex", func(t *testing.T) {
|
|
// Try CreateIndex for testindex1
|
|
_, err := gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: "testindex1", Keys: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
schema, err := m.API.Schema(ctx, false)
|
|
if err != nil {
|
|
t.Fatal("Getting schema error", err)
|
|
}
|
|
if len(schema) != 1 {
|
|
t.Fatal("Schema should include one index")
|
|
}
|
|
if schema[0].Name != "testindex1" {
|
|
t.Fatal("Index name not set correctly")
|
|
}
|
|
if schema[0].Options.Keys != true {
|
|
t.Fatal("Index Keys not set correctly")
|
|
}
|
|
if schema[0].Options.TrackExistence != true {
|
|
t.Fatal("Index TrackExistence should be true when created by gRPC")
|
|
}
|
|
|
|
// Try CreateIndex for testindex2
|
|
_, err = gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: "testindex2"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
schema, err = m.API.Schema(ctx, false)
|
|
if err != nil {
|
|
t.Fatal("Getting schema error", err)
|
|
}
|
|
|
|
if len(schema) != 2 {
|
|
t.Fatal("Schema should include two indexes")
|
|
}
|
|
|
|
_ = m.API.DeleteIndex(ctx, "testindex1")
|
|
|
|
schema, err = m.API.Schema(ctx, false)
|
|
if err != nil {
|
|
t.Fatal("Getting schema error", err)
|
|
}
|
|
|
|
if len(schema) != 1 {
|
|
t.Fatal("Schema should include one index")
|
|
}
|
|
if schema[0].Name != "testindex2" {
|
|
t.Fatal("Index name not set correctly")
|
|
}
|
|
if schema[0].Options.Keys != false {
|
|
t.Fatal("Index Keys not set correctly")
|
|
}
|
|
|
|
// Check errors for CreateIndex: create index with same name
|
|
_, err = gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: "testindex2"})
|
|
errStatus, _ := status.FromError(err)
|
|
if errStatus.Code() != codes.AlreadyExists {
|
|
t.Fatalf("Error code should be codes.AlreadyExists, but is %v", errStatus.Code())
|
|
}
|
|
|
|
// Check errors for CreateIndex: create index with no name
|
|
_, err = gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: ""})
|
|
errStatus, _ = status.FromError(err)
|
|
if errStatus.Code() != codes.FailedPrecondition {
|
|
t.Fatalf("Error code should be codes.Unknown, but is %v", errStatus.Code())
|
|
}
|
|
|
|
// Check errors for CreateIndex: create index with invalid name
|
|
_, err = gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: "💩"})
|
|
errStatus, _ = status.FromError(err)
|
|
if errStatus.Code() != codes.FailedPrecondition {
|
|
t.Fatalf("Error code should be codes.FailedPrecondition, but is %v", errStatus.Code())
|
|
}
|
|
|
|
_ = m.API.DeleteIndex(ctx, "testindex2")
|
|
})
|
|
|
|
t.Run("GetIndex", func(t *testing.T) {
|
|
_, err := m.API.CreateIndex(ctx, "testindex1", pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Check GetIndex for testindex1
|
|
resp, err := gh.GetIndex(ctx, &pb.GetIndexRequest{Name: "testindex1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if resp.Index.Name != "testindex1" {
|
|
t.Fatalf("Index name does not match: %s", resp.Index.Name)
|
|
}
|
|
|
|
// Check errors for GetIndex: get index that doesn't exist
|
|
_, err = gh.GetIndex(ctx, &pb.GetIndexRequest{Name: "wrongname"})
|
|
errStatus, _ := status.FromError(err)
|
|
if errStatus.Code() != codes.NotFound {
|
|
t.Fatalf("Error code should be codes.NotFound, but is %v", errStatus.Code())
|
|
}
|
|
|
|
// Check errors for GetIndex: get index with invalid name
|
|
_, err = gh.GetIndex(ctx, &pb.GetIndexRequest{Name: "💩"})
|
|
errStatus, _ = status.FromError(err)
|
|
if errStatus.Code() != codes.NotFound {
|
|
t.Fatalf("Error code should be codes.NotFound, but is %v", errStatus.Code())
|
|
}
|
|
_ = m.API.DeleteIndex(ctx, "testindex1")
|
|
})
|
|
|
|
t.Run("GetIndexes", func(t *testing.T) {
|
|
_, err := m.API.CreateIndex(ctx, "testindex1", pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Check GetIndexes
|
|
resp2, err := gh.GetIndexes(ctx, &pb.GetIndexesRequest{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(resp2.Indexes) != 1 && resp2.Indexes[0].Name != "testindex1" {
|
|
t.Fatalf("GetIndexes did not produce the correct result set: %v", resp2.Indexes)
|
|
}
|
|
|
|
_, err = m.API.CreateIndex(ctx, "testindex2", pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Check GetIndexes again
|
|
resp, err := gh.GetIndexes(ctx, &pb.GetIndexesRequest{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(resp.Indexes) != 2 {
|
|
t.Fatalf("GetIndexes did not produce the correct result set: %v", resp.Indexes)
|
|
}
|
|
_ = m.API.DeleteIndex(ctx, "testindex1")
|
|
_ = m.API.DeleteIndex(ctx, "testindex2")
|
|
})
|
|
|
|
t.Run("DeleteIndexes", func(t *testing.T) {
|
|
_, err := m.API.CreateIndex(ctx, "testindex1", pilosa.IndexOptions{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Try to delete index
|
|
_, err = gh.DeleteIndex(ctx, &pb.DeleteIndexRequest{Name: "testindex1"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
schema, err := m.API.Schema(ctx, false)
|
|
if err != nil {
|
|
t.Fatal("Getting schema error", err)
|
|
}
|
|
|
|
if len(schema) != 0 {
|
|
t.Fatal("Schema should include no index")
|
|
}
|
|
|
|
// Try to delete non-existing index
|
|
_, err = gh.DeleteIndex(ctx, &pb.DeleteIndexRequest{Name: "doesnotexist"})
|
|
errStatus, _ := status.FromError(err)
|
|
if errStatus.Code() != codes.NotFound {
|
|
t.Fatalf("Error code should be codes.NotFound, but is %v", errStatus.Code())
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestLogQuery(t *testing.T) {
|
|
method := "test!"
|
|
uinfo := &authn.UserInfo{
|
|
UserID: "ID",
|
|
UserName: "name",
|
|
}
|
|
ctx := authn.WithUserInfo(context.Background(), uinfo)
|
|
|
|
cases := []struct {
|
|
name string
|
|
req interface{}
|
|
expected string
|
|
}{
|
|
{
|
|
name: "nonQueryReq",
|
|
req: "nope",
|
|
expected: fmt.Sprintf("GRPC: %v, %v, %v, %v, %v\n", "", []string{}, "test!", uinfo.UserID, uinfo.UserName),
|
|
},
|
|
{
|
|
name: "QuerySQLReq",
|
|
req: &pb.QuerySQLRequest{Sql: "show fields from table"},
|
|
expected: fmt.Sprintf("GRPC: %v, %v, %v, %v, %v, %v\n", "", []string{}, "test!", uinfo.UserID, uinfo.UserName, "show fields from table"),
|
|
},
|
|
{
|
|
name: "QueryPQLReq",
|
|
req: &pb.QueryPQLRequest{Index: "index", Pql: "Count(All())"},
|
|
expected: fmt.Sprintf("GRPC: %v, %v, %v, %v, %v, %v\n", "", []string{}, "test!", uinfo.UserID, uinfo.UserName, "[index]Count(All())"),
|
|
},
|
|
}
|
|
for _, test := range cases {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
buf := new(bytes.Buffer)
|
|
l := logger.NewStandardLogger(buf)
|
|
server.LogQuery(ctx, method, test.req, l)
|
|
if !strings.HasSuffix(buf.String(), test.expected) {
|
|
t.Errorf("expected '%v', got '%v'", test.expected, buf.String())
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func setUpTestQuerySQLUnary(ctx context.Context, t *testing.T) (gh *server.GRPCHandler, tearDownFunc func()) {
|
|
t.Helper()
|
|
|
|
m := test.RunCommand(t)
|
|
gh = server.NewGRPCHandler(m.API).WithQueryLogger(logger.NewStandardLogger(os.Stdout))
|
|
// grouper
|
|
grouper := m.MustCreateIndex(t, "grouper", pilosa.IndexOptions{Keys: false, TrackExistence: true})
|
|
m.MustCreateField(t, grouper.Name(), "color", pilosa.OptFieldKeys())
|
|
for id, color := range map[int]string{
|
|
1: "blue",
|
|
2: "blue",
|
|
5: "blue",
|
|
6: "blue",
|
|
7: "blue",
|
|
3: "red",
|
|
9: "red",
|
|
10: "red",
|
|
4: "green",
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: grouper.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, color="%s")`, id, color),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
m.MustCreateField(t, grouper.Name(), "score", pilosa.OptFieldTypeInt(-1000, 1000))
|
|
for id, score := range map[int]int{
|
|
1: -10,
|
|
2: -8,
|
|
3: 6,
|
|
4: 0,
|
|
5: -2,
|
|
6: 100,
|
|
7: 0,
|
|
8: -13,
|
|
9: 80,
|
|
10: -2,
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: grouper.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, score=%d)`, id, score),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
m.MustCreateField(t, grouper.Name(), "age", pilosa.OptFieldTypeInt(0, 100))
|
|
for id, age := range map[int]int{
|
|
2: 16,
|
|
5: 16,
|
|
8: 16,
|
|
9: 16,
|
|
3: 19,
|
|
1: 27,
|
|
4: 27,
|
|
7: 27,
|
|
10: 31,
|
|
6: 34,
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: grouper.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, age=%d)`, id, age),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
m.MustCreateField(t, grouper.Name(), "height", pilosa.OptFieldTypeInt(0, 1000))
|
|
for id, height := range map[int]int{
|
|
1: 20,
|
|
2: 30,
|
|
3: 40,
|
|
4: 50,
|
|
5: 60,
|
|
6: 70,
|
|
7: 80,
|
|
8: 90,
|
|
9: 100,
|
|
10: 110,
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: grouper.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, height=%d)`, id, height),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
m.MustCreateField(t, grouper.Name(), "timestamp", pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds))
|
|
for id, timestamp := range map[int]string{
|
|
1: "2011-04-02T12:32:00Z",
|
|
2: "2011-01-02T12:32:00Z",
|
|
3: "2012-01-02T12:32:00Z",
|
|
4: "2013-09-02T12:32:00Z",
|
|
5: "2014-01-02T12:32:00Z",
|
|
6: "2010-05-02T12:32:00Z",
|
|
7: "2016-08-02T12:32:00Z",
|
|
8: "2020-01-02T12:32:00Z",
|
|
9: "2000-03-02T12:32:00Z",
|
|
10: "2018-01-02T12:32:00Z",
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: grouper.Name(),
|
|
Pql: fmt.Sprintf("Set(%d, timestamp=\"%s\")", id, timestamp),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// joiner
|
|
joiner := m.MustCreateIndex(t, "joiner", pilosa.IndexOptions{TrackExistence: true})
|
|
m.MustCreateField(t, joiner.Name(), "grouperid", pilosa.OptFieldTypeInt(0, 1000), pilosa.OptFieldForeignIndex(grouper.Name()))
|
|
m.MustCreateField(t, joiner.Name(), "jointype", pilosa.OptFieldTypeInt(-1000, 1000))
|
|
for id, grouperid := range map[int]int{
|
|
1: 1,
|
|
2: 2,
|
|
3: 5,
|
|
4: 6,
|
|
5: 7,
|
|
6: 3,
|
|
7: 8,
|
|
8: 9,
|
|
9: 1,
|
|
10: 2,
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: joiner.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, grouperid=%d)`, id, grouperid),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
for id, jointype := range map[int]int{
|
|
1: 1,
|
|
2: 1,
|
|
3: 1,
|
|
4: 1,
|
|
5: 1,
|
|
6: 2,
|
|
7: 2,
|
|
8: 2,
|
|
9: 3,
|
|
10: 3,
|
|
} {
|
|
if _, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{
|
|
Index: joiner.Name(),
|
|
Pql: fmt.Sprintf(`Set(%d, jointype=%d)`, id, jointype),
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// delete_me
|
|
m.MustCreateIndex(t, "delete_me", pilosa.IndexOptions{TrackExistence: true})
|
|
|
|
m.MustCreateIndex(t, "another_one", pilosa.IndexOptions{TrackExistence: true})
|
|
m.MustCreateIndex(t, "deletable_index", pilosa.IndexOptions{TrackExistence: true})
|
|
|
|
return gh, func() {
|
|
if err := m.API.DeleteIndex(ctx, joiner.Name()); err != nil {
|
|
panic(err)
|
|
}
|
|
if err := m.API.DeleteIndex(ctx, grouper.Name()); err != nil {
|
|
panic(err)
|
|
}
|
|
if err := m.Close(); err != nil {
|
|
panic(err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func toTableResponse(resp *pb.TableResponse) tableResponse {
|
|
tr := tableResponse{
|
|
headers: make([]columnInfo, len(resp.Headers)),
|
|
rows: make([]row, len(resp.Rows)),
|
|
}
|
|
|
|
for i, h := range resp.Headers {
|
|
tr.headers[i] = columnInfo{
|
|
name: h.Name,
|
|
datatype: h.Datatype,
|
|
}
|
|
}
|
|
|
|
for i, r := range resp.Rows {
|
|
tr.rows[i].columns = make([]columnResponse, len(r.Columns))
|
|
for j, c := range r.Columns {
|
|
|
|
switch v := c.GetColumnVal().(type) {
|
|
case *pb.ColumnResponse_StringVal:
|
|
tr.rows[i].columns[j] = v.StringVal
|
|
case *pb.ColumnResponse_Uint64Val:
|
|
tr.rows[i].columns[j] = v.Uint64Val
|
|
case *pb.ColumnResponse_Int64Val:
|
|
tr.rows[i].columns[j] = v.Int64Val
|
|
case *pb.ColumnResponse_BoolVal:
|
|
tr.rows[i].columns[j] = v.BoolVal
|
|
case *pb.ColumnResponse_BlobVal:
|
|
tr.rows[i].columns[j] = v.BlobVal
|
|
case *pb.ColumnResponse_Uint64ArrayVal:
|
|
tr.rows[i].columns[j] = v.Uint64ArrayVal.Vals
|
|
case *pb.ColumnResponse_StringArrayVal:
|
|
tr.rows[i].columns[j] = v.StringArrayVal.Vals
|
|
case *pb.ColumnResponse_Float64Val:
|
|
tr.rows[i].columns[j] = v.Float64Val
|
|
case *pb.ColumnResponse_DecimalVal:
|
|
tr.rows[i].columns[j] = pql.NewDecimal(v.DecimalVal.Value, v.DecimalVal.Scale)
|
|
case *pb.ColumnResponse_TimestampVal:
|
|
tr.rows[i].columns[j] = v.TimestampVal
|
|
default:
|
|
tr.rows[i].columns[j] = nil
|
|
}
|
|
}
|
|
}
|
|
|
|
return tr
|
|
}
|
|
|
|
func equal(exp tableResponse, got tableResponse) error {
|
|
if !reflect.DeepEqual(exp, got) {
|
|
return fmt.Errorf("got: %+v %[1]T, but expected: %+v", got, exp)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func equalUnordered(exp tableResponse, got tableResponse) error {
|
|
if len(exp.headers) != len(got.headers) || !reflect.DeepEqual(exp.headers, got.headers) {
|
|
return fmt.Errorf("header does not match: got %+v, but expected %+v", got.headers, exp.headers)
|
|
}
|
|
|
|
if len(exp.rows) != len(got.rows) {
|
|
return fmt.Errorf("rows count does not match: got %+v, but expected %+v", len(got.rows), len(exp.rows))
|
|
}
|
|
for _, er := range exp.rows {
|
|
for j, gr := range got.rows {
|
|
if reflect.DeepEqual(er.columns, gr.columns) {
|
|
got.rows[j] = got.rows[len(got.rows)-1]
|
|
got.rows = got.rows[:len(got.rows)-1]
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if len(got.rows) > 0 {
|
|
return fmt.Errorf("got incorrect rows: %+v", got.rows)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type MockServerTransportStream struct {
|
|
header metadata.MD
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) Method() string {
|
|
return ""
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) SetHeader(md metadata.MD) error {
|
|
// Should probably merge md with value of stream.header, but this works since we have only one metadata value
|
|
stream.header = md
|
|
return nil
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) SendHeader(md metadata.MD) error {
|
|
stream.header = md
|
|
return nil
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) SetTrailer(md metadata.MD) error {
|
|
return nil
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) GetDuration() (int, error) {
|
|
duration, ok := stream.header["duration"]
|
|
if ok {
|
|
return strconv.Atoi(duration[0])
|
|
}
|
|
return 0, errors.New("duration not recorded")
|
|
}
|
|
|
|
func (stream *MockServerTransportStream) ClearMD() {
|
|
stream.header = metadata.New(map[string]string{})
|
|
}
|
|
|
|
type mockPilosa_QuerySQLServer struct {
|
|
MockServerTransportStream
|
|
ctx context.Context
|
|
pb.Pilosa_QuerySQLServer
|
|
Results []*pb.RowResponse
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) Send(result *pb.RowResponse) error {
|
|
m.Results = append(m.Results, result)
|
|
return nil
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) SendHeader(md metadata.MD) error {
|
|
return m.MockServerTransportStream.SendHeader(md)
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) SetHeader(md metadata.MD) error {
|
|
return m.MockServerTransportStream.SetHeader(md)
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) SetTrailer(md metadata.MD) {
|
|
_ = m.MockServerTransportStream.SetTrailer(md)
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) Context() context.Context {
|
|
return m.ctx
|
|
}
|
|
|
|
func (m *mockPilosa_QuerySQLServer) clearResults() {
|
|
m.Results = m.Results[:0]
|
|
}
|
|
func writeTestFile(t *testing.T, filename, content string) string {
|
|
fname := filepath.Join(t.TempDir(), filename)
|
|
f, err := os.Create(fname)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, err = io.WriteString(f, content)
|
|
assert.NoError(t, err)
|
|
if err := f.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return fname
|
|
}
|
|
|
|
func Test_ChainUnaryInterceptor(t *testing.T) {
|
|
|
|
salt := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
|
|
r := req.(*pb.QueryPQLRequest)
|
|
r.Pql = fmt.Sprintf("%s, add salt", r.Pql)
|
|
return handler(ctx, r)
|
|
|
|
}
|
|
pepper := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
|
|
r := req.(*pb.QueryPQLRequest)
|
|
r.Pql = fmt.Sprintf("%s, add pepper", r.Pql)
|
|
return handler(ctx, r)
|
|
|
|
}
|
|
|
|
interceptors0 := []grpc.UnaryServerInterceptor{}
|
|
interceptors1 := []grpc.UnaryServerInterceptor{salt}
|
|
interceptors2 := []grpc.UnaryServerInterceptor{salt, pepper}
|
|
|
|
tests := []struct {
|
|
name string
|
|
interceptors []grpc.UnaryServerInterceptor
|
|
want string
|
|
}{
|
|
{"0", interceptors0, "Soup"},
|
|
{"1", interceptors1, "Soup, add salt"},
|
|
{"2", interceptors2, "Soup, add salt, add pepper"},
|
|
}
|
|
for _, tt := range tests {
|
|
ctx := context.Background()
|
|
req := &pb.QueryPQLRequest{
|
|
Pql: "Soup",
|
|
}
|
|
info := &grpc.UnaryServerInfo{}
|
|
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
|
|
r := req.(*pb.QueryPQLRequest)
|
|
return r.Pql, nil
|
|
}
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
r := req
|
|
chained := server.ChainUnaryInterceptor(tt.interceptors...)
|
|
resp, err := chained(ctx, r, info, handler)
|
|
vprint.VV("resp: %+v", resp)
|
|
|
|
if err != nil {
|
|
t.Errorf("ChainUnaryInterceptor() error = %v", err)
|
|
}
|
|
if resp != tt.want {
|
|
t.Errorf("ChainUnaryInterceptor() = %v, want %v", resp, tt.want)
|
|
}
|
|
|
|
})
|
|
}
|
|
}
|
|
|
|
// Testing object that implements the ServerStream interface
|
|
type MockStream struct {
|
|
context context.Context
|
|
}
|
|
|
|
func (ms *MockStream) SetHeader(md metadata.MD) error {
|
|
ms.context = contextWithMetadata(context.Background(), md)
|
|
return nil
|
|
}
|
|
|
|
func (ms MockStream) SendHeader(metadata.MD) error {
|
|
return nil
|
|
}
|
|
|
|
func (ms MockStream) SetTrailer(metadata.MD) {}
|
|
|
|
func (ms MockStream) Context() context.Context {
|
|
if ms.context == nil {
|
|
md := metadata.New(map[string]string{})
|
|
ms.context = contextWithMetadata(context.Background(), md)
|
|
}
|
|
return ms.context
|
|
}
|
|
|
|
func (ms MockStream) SendMsg(m interface{}) error {
|
|
return nil
|
|
}
|
|
|
|
func (ms MockStream) RecvMsg(m interface{}) error {
|
|
return nil
|
|
}
|
|
|
|
type contextKeyMetadata struct{}
|
|
|
|
func contextWithMetadata(ctx context.Context, metadata metadata.MD) context.Context {
|
|
return context.WithValue(ctx, contextKeyMetadata{}, metadata)
|
|
}
|
|
|
|
func metadataFromContext(ctx context.Context) (meta metadata.MD, ok bool) {
|
|
meta, ok = ctx.Value(contextKeyMetadata{}).(metadata.MD)
|
|
return
|
|
}
|
|
|
|
func Test_ChainStreamInterceptor(t *testing.T) {
|
|
|
|
salt := func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
md, ok := metadataFromContext(ss.Context())
|
|
if !ok {
|
|
t.Fatal("metadata was not in context")
|
|
}
|
|
md.Append("ingredient", "with salt")
|
|
err := ss.SetHeader(md)
|
|
assert.NoError(t, err)
|
|
return handler(srv, ss)
|
|
}
|
|
pepper := func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
|
md, ok := metadataFromContext(ss.Context())
|
|
if !ok {
|
|
t.Fatal("metadata was not in context")
|
|
}
|
|
md.Append("ingredient", "and pepper")
|
|
err := ss.SetHeader(md)
|
|
assert.NoError(t, err)
|
|
return handler(srv, ss)
|
|
}
|
|
|
|
interceptors0 := []grpc.StreamServerInterceptor{}
|
|
interceptors1 := []grpc.StreamServerInterceptor{salt}
|
|
interceptors2 := []grpc.StreamServerInterceptor{salt, pepper}
|
|
|
|
tests := []struct {
|
|
name string
|
|
interceptors []grpc.StreamServerInterceptor
|
|
want []string
|
|
}{
|
|
{"0", interceptors0, []string{"Soup"}},
|
|
{"1", interceptors1, []string{"Soup", "with salt"}},
|
|
{"2", interceptors2, []string{"Soup", "with salt", "and pepper"}},
|
|
}
|
|
for _, tt := range tests {
|
|
result := make([]string, 0)
|
|
srv := "asdf"
|
|
md := metadata.New(map[string]string{})
|
|
ss := &MockStream{context: contextWithMetadata(context.Background(), md)}
|
|
info := &grpc.StreamServerInfo{}
|
|
handler := func(srv interface{}, stream grpc.ServerStream) error {
|
|
md, ok := metadataFromContext(stream.Context())
|
|
if !ok {
|
|
t.Fatal("metadata was not in context")
|
|
}
|
|
vals := md.Get("ingredient")
|
|
result = append(result, "Soup")
|
|
result = append(result, vals...)
|
|
return nil
|
|
}
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
chained := server.ChainStreamInterceptors(tt.interceptors...)
|
|
err := chained(srv, ss, info, handler)
|
|
assert.NoError(t, err)
|
|
if !reflect.DeepEqual(result, tt.want) {
|
|
t.Errorf("ChainStreamInterceptor() = %v, want %v", result, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|