mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
add comments, simplify tests, move ToCSV code
generally, address code review feedback
This commit is contained in:
Matt Jaffee
parent
27ca9dab36
commit
757c8a86b5
4 changed files with 138 additions and 144 deletions
32
executor.go
32
executor.go
|
|
@ -563,7 +563,6 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
|
|||
} else {
|
||||
v, err = e.executeCall(ctx, qcx, index, call, shards, opt)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -6028,13 +6027,26 @@ func (e *executor) translateResults(ctx context.Context, index string, idx *Inde
|
|||
return nil
|
||||
}
|
||||
|
||||
// translationStrategy denotes the several different ways the bits in
|
||||
// a *Row could be translated to string keys.
|
||||
type translationStrategy int
|
||||
|
||||
const (
|
||||
// byCurrentIndex means to interpret the bits as IDs in "top
|
||||
// level" index for this query (e.g. the index specified in the
|
||||
// path of the HTTP request).
|
||||
byCurrentIndex translationStrategy = iota + 1
|
||||
// byRowField means that the bits in this *Row are row IDs which
|
||||
// should be translated using the field's (*Row.Field) translation store.
|
||||
byRowField
|
||||
// byRowFieldForeignIndex means that the bits in this *Row should
|
||||
// be interpreted as IDs in the foreign index of the *Row.Field.
|
||||
byRowFieldForeignIndex
|
||||
// byRowIndex means the bits in this *Row should be translated
|
||||
// according to the index named by *Row.Index
|
||||
byRowIndex
|
||||
// noTranslation means the bits should not be translated to string
|
||||
// keys.
|
||||
noTranslation
|
||||
)
|
||||
|
||||
|
|
@ -6063,7 +6075,7 @@ func (e *executor) howToTranslate(idx *Index, row *Row) (rowIdx *Index, rowField
|
|||
|
||||
// Handle the case where the Row has specified a field.
|
||||
if rowField != nil {
|
||||
// Handle case where field has a foreign index.
|
||||
// Handle the case where field has a foreign index.
|
||||
if rowField.ForeignIndex() != "" {
|
||||
fidx := e.Holder.Index(rowField.ForeignIndex())
|
||||
if fidx == nil {
|
||||
|
|
@ -6084,7 +6096,7 @@ func (e *executor) howToTranslate(idx *Index, row *Row) (rowIdx *Index, rowField
|
|||
return rowIdx, rowField, byRowIndex, nil
|
||||
}
|
||||
|
||||
// Handle normal case (row represents a set of records in
|
||||
// Handle the normal case (row represents a set of records in
|
||||
// the top level index, Row has not specifed a different index
|
||||
// or field).
|
||||
if rowIdx == idx && idx.Keys() && rowField == nil {
|
||||
|
|
@ -6096,20 +6108,18 @@ func (e *executor) howToTranslate(idx *Index, row *Row) (rowIdx *Index, rowField
|
|||
func (e *executor) collectResultIDs(index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]struct{}) error {
|
||||
switch result := result.(type) {
|
||||
case *Row:
|
||||
// Only collect result IDs if they are in the current index.
|
||||
_, _, strategy, err := e.howToTranslate(idx, result)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "determining how to translate")
|
||||
}
|
||||
// Only collect result IDs if they are in the current index.
|
||||
if strategy != byCurrentIndex {
|
||||
return nil
|
||||
}
|
||||
for _, segment := range result.Segments() {
|
||||
for _, col := range segment.Columns() {
|
||||
idSet[col] = struct{}{}
|
||||
if strategy == byCurrentIndex {
|
||||
for _, segment := range result.Segments() {
|
||||
for _, col := range segment.Columns() {
|
||||
idSet[col] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
case ExtractedIDMatrix:
|
||||
for _, col := range result.Columns {
|
||||
idSet[col.ColumnID] = struct{}{}
|
||||
|
|
|
|||
192
executor_test.go
192
executor_test.go
|
|
@ -17,9 +17,11 @@ package pilosa_test
|
|||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"math"
|
||||
"math/rand"
|
||||
|
|
@ -37,6 +39,7 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/boltdb"
|
||||
"github.com/pilosa/pilosa/v2/http"
|
||||
"github.com/pilosa/pilosa/v2/pql"
|
||||
"github.com/pilosa/pilosa/v2/proto"
|
||||
"github.com/pilosa/pilosa/v2/server"
|
||||
"github.com/pilosa/pilosa/v2/test"
|
||||
"github.com/pilosa/pilosa/v2/testhook"
|
||||
|
|
@ -6799,7 +6802,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
tests := []struct {
|
||||
query string
|
||||
qrVerifier func(t *testing.T, resp pilosa.QueryResponse)
|
||||
csvVerifier func(t *testing.T, resp string)
|
||||
csvVerifier string
|
||||
}{
|
||||
{
|
||||
query: "Count(All())",
|
||||
|
|
@ -6808,12 +6811,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("expected 6, got %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "6\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "6\n",
|
||||
},
|
||||
{
|
||||
query: "Count(Distinct(field=likenums))",
|
||||
|
|
@ -6822,12 +6820,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong count: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "7\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "7\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(field=likenums)",
|
||||
|
|
@ -6836,12 +6829,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v %+v", resp.Results[0].(*pilosa.Row).Columns(), resp.Results[0].(*pilosa.Row))
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "1\n2\n3\n4\n5\n6\n7\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "1\n2\n3\n4\n5\n6\n7\n",
|
||||
},
|
||||
{
|
||||
query: "Count(Distinct(field=likes))",
|
||||
|
|
@ -6850,12 +6838,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong count: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "7\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "7\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(field=affinity)",
|
||||
|
|
@ -6867,12 +6850,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong negative records: %+v", resp.Results[0].(pilosa.SignedRow).Neg.Columns())
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "-10\n-5\n0\n5\n10\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "-10\n-5\n0\n5\n10\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(Row(affinity>=0),field=affinity)",
|
||||
|
|
@ -6884,12 +6862,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong negative records: %+v", resp.Results[0].(pilosa.SignedRow).Neg.Columns())
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "0\n5\n10\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "0\n5\n10\n",
|
||||
},
|
||||
{
|
||||
query: "Count(Distinct(Row(affinity>=0),field=affinity))",
|
||||
|
|
@ -6898,12 +6871,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong number of values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "3\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "3\n",
|
||||
},
|
||||
|
||||
// Handling this case properly will require changing the way
|
||||
|
|
@ -6927,12 +6895,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "pilosa\nzebra\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "pilosa\nzebra\nicecream\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(Row(affinity>0),field=likes)",
|
||||
|
|
@ -6941,12 +6904,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "molecula\npangolin\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "molecula\npangolin\nicecream\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(Row(likenums=1),field=likes)",
|
||||
|
|
@ -6955,12 +6913,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "molecula\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "molecula\nicecream\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(field=likes)",
|
||||
|
|
@ -6969,12 +6922,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(All(),field=likes)",
|
||||
|
|
@ -6983,12 +6931,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n",
|
||||
},
|
||||
{
|
||||
query: "Distinct(field=likes )",
|
||||
|
|
@ -6997,12 +6940,7 @@ func TestVariousQueries(t *testing.T) {
|
|||
t.Errorf("wrong values: %+v", resp.Results[0])
|
||||
}
|
||||
},
|
||||
csvVerifier: func(t *testing.T, resp string) {
|
||||
exp := "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n"
|
||||
if resp != exp {
|
||||
t.Errorf("expected '%s', got '%s'", exp, resp)
|
||||
}
|
||||
},
|
||||
csvVerifier: "molecula\npilosa\npangolin\nzebra\ntoucan\ndog\nicecream\n",
|
||||
},
|
||||
}
|
||||
|
||||
|
|
@ -7013,12 +6951,104 @@ func TestVariousQueries(t *testing.T) {
|
|||
if tst.qrVerifier != nil {
|
||||
tst.qrVerifier(t, resp)
|
||||
}
|
||||
csvString := tr.ToCSVString()
|
||||
csvString, err := tableResponseToCSVString(tr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// verify everything after header
|
||||
tst.csvVerifier(t, csvString[strings.Index(csvString, "\n")+1:])
|
||||
got := csvString[strings.Index(csvString, "\n")+1:]
|
||||
if got != tst.csvVerifier {
|
||||
t.Errorf("expected '%s', got '%s'", tst.csvVerifier, got)
|
||||
}
|
||||
|
||||
// TODO: add HTTP and Postgres and ability to convert
|
||||
// those results to CSV to run through CSV verifier
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestReproDistinctWFilterIssue(t *testing.T) {
|
||||
|
||||
c := test.MustRunCluster(t, 3)
|
||||
defer c.Close()
|
||||
r := rand.New(rand.NewSource(127))
|
||||
|
||||
for i := 0; i < 77; i++ {
|
||||
index := fmt.Sprintf("users%d", i)
|
||||
c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likes", pilosa.OptFieldKeys())
|
||||
c.CreateField(t, index, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "filterfield", pilosa.OptFieldKeys())
|
||||
likes := make([][2]string, 0)
|
||||
filter := make([][2]string, 0)
|
||||
for userNum := 0; userNum < 100; userNum++ {
|
||||
likes = append(likes, [2]string{fmt.Sprintf("like%d", r.Intn(95)), "user" + strconv.Itoa(userNum)})
|
||||
if r.Intn(10) < 8 {
|
||||
filter = append(filter, [2]string{"yes", "user" + strconv.Itoa(userNum)})
|
||||
}
|
||||
}
|
||||
c.ImportKeyKey(t, index, "likes", likes)
|
||||
c.ImportKeyKey(t, index, "filterfield", filter)
|
||||
|
||||
distinctRes := c.Query(t, index, "Count(Distinct(Row(filterfield=yes), field=likes))")
|
||||
distinctCount := distinctRes.Results[0].(uint64)
|
||||
groupbyRes := c.Query(t, index, "GroupBy(Rows(field=likes), filter=Row(filterfield=yes))")
|
||||
groupbyCount := uint64(len(groupbyRes.Results[0].([]pilosa.GroupCount)))
|
||||
|
||||
t.Logf("D:%v", distinctRes.Results[0])
|
||||
t.Logf("G:%v", groupbyRes.Results[0])
|
||||
if distinctCount != groupbyCount {
|
||||
t.Errorf("distinct: %d, groupby: %d", distinctCount, groupbyCount)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tableResponseToCSV converts a generic TableResponse to a CSV format
|
||||
// and writes it to the writer.
|
||||
func tableResponseToCSV(m *proto.TableResponse, w io.Writer) error {
|
||||
writer := csv.NewWriter(w)
|
||||
record := make([]string, len(m.Headers))
|
||||
for i, h := range m.Headers {
|
||||
record[i] = h.Name
|
||||
}
|
||||
err := writer.Write(record)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "writing header")
|
||||
}
|
||||
for i, row := range m.Rows {
|
||||
record = record[:0]
|
||||
for colIndex, col := range row.Columns {
|
||||
switch m.Headers[colIndex].Datatype {
|
||||
case "[]string":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetStringArrayVal()))
|
||||
case "[]uint64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetUint64ArrayVal()))
|
||||
case "string":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetStringVal()))
|
||||
case "uint64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetUint64Val()))
|
||||
case "decimal":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetDecimalVal().String()))
|
||||
case "bool":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetBoolVal()))
|
||||
case "int64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetInt64Val()))
|
||||
}
|
||||
}
|
||||
err := writer.Write(record)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "writing row %d", i)
|
||||
}
|
||||
}
|
||||
writer.Flush()
|
||||
return errors.Wrap(writer.Error(), "writing or flushing CSV")
|
||||
}
|
||||
|
||||
// tableResponseToCSVString converts a generic TableResponse to a CSV format
|
||||
// and returns it as a string.
|
||||
func tableResponseToCSVString(m *proto.TableResponse) (string, error) {
|
||||
buf := &bytes.Buffer{}
|
||||
err := tableResponseToCSV(m, buf)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "writing tableResponse CSV to bytes.Buffer")
|
||||
}
|
||||
return buf.String(), nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,8 +15,6 @@
|
|||
package proto
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/csv"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
|
|
@ -327,51 +325,3 @@ func (c ConstRowser) ToRows(fn func(*RowResponse) error) error {
|
|||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *TableResponse) ToCSV(w io.Writer) error {
|
||||
writer := csv.NewWriter(w)
|
||||
record := make([]string, len(m.Headers))
|
||||
for i, h := range m.Headers {
|
||||
record[i] = h.Name
|
||||
}
|
||||
err := writer.Write(record)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "writing header")
|
||||
}
|
||||
for i, row := range m.Rows {
|
||||
record = record[:0]
|
||||
for colIndex, col := range row.Columns {
|
||||
switch m.Headers[colIndex].Datatype {
|
||||
case "[]string":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetStringArrayVal()))
|
||||
case "[]uint64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetUint64ArrayVal()))
|
||||
case "string":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetStringVal()))
|
||||
case "uint64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetUint64Val()))
|
||||
case "decimal":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetDecimalVal().String()))
|
||||
case "bool":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetBoolVal()))
|
||||
case "int64":
|
||||
record = append(record, fmt.Sprintf("%v", col.GetInt64Val()))
|
||||
}
|
||||
}
|
||||
err := writer.Write(record)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "writing row %d", i)
|
||||
}
|
||||
}
|
||||
writer.Flush()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *TableResponse) ToCSVString() string {
|
||||
buf := &bytes.Buffer{}
|
||||
err := m.ToCSV(buf)
|
||||
if err != nil {
|
||||
panic(fmt.Sprintf("shouldn't get an error writing to bytes.Buffer, got: %v", err))
|
||||
}
|
||||
return buf.String()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -55,6 +55,9 @@ func (c *Cluster) Query(t testing.TB, index, query string) pilosa.QueryResponse
|
|||
return c.Nodes[0].QueryAPI(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
}
|
||||
|
||||
// QueryHTTP executes a PQL query through the HTTP endpoint. It fails
|
||||
// the test for explicit errors, but returns an error which has the
|
||||
// response body if the HTTP call returns a non-OK status.
|
||||
func (c *Cluster) QueryHTTP(t testing.TB, index, query string) (string, error) {
|
||||
t.Helper()
|
||||
if len(c.Nodes) == 0 {
|
||||
|
|
@ -63,6 +66,8 @@ func (c *Cluster) QueryHTTP(t testing.TB, index, query string) (string, error) {
|
|||
return c.Nodes[0].Query(t, index, "", query)
|
||||
}
|
||||
|
||||
// QueryGRPC executes a PQL query through the GRPC endpoint. It fails the
|
||||
// test if there is an error.
|
||||
func (c *Cluster) QueryGRPC(t testing.TB, index, query string) *proto.TableResponse {
|
||||
t.Helper()
|
||||
if len(c.Nodes) == 0 {
|
||||
|
|
@ -190,8 +195,7 @@ type KeyID struct {
|
|||
ID uint64
|
||||
}
|
||||
|
||||
// ImportIDKey imports data into an index where the index is using
|
||||
// keys, but the field is not.
|
||||
//ImportIDKey imports data into an unkeyed set field in a keyed index.
|
||||
func (c *Cluster) ImportIDKey(t testing.TB, index, field string, pairs []KeyID) {
|
||||
t.Helper()
|
||||
importRequest := &pilosa.ImportRequest{
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue