scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

ToTable and ToRows interface for gRPC

Browse source
This commit is contained in:
Travis 2020-04-11 15:23:32 -05:00
parent 0691b6c015
commit c45a4bf3dc
7 changed files with 571 additions and 351 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

86
cache.go
View file

@ -24,7 +24,9 @@ import (
"time"
"github.com/pilosa/pilosa/v2/lru"
pb "github.com/pilosa/pilosa/v2/proto"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pkg/errors"
)
const (
@ -323,12 +325,44 @@ type Pair struct {
Count uint64 `json:"count"`
}
// PairField
// PairField is a Pair with its associated field.
type PairField struct {
Pair Pair
Field string
}
// ToTable implements the ToTabler interface.
func (p PairField) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(p, 1)
}
// ToRows implements the ToRowser interface.
func (p PairField) ToRows(callback func(*pb.RowResponse) error) error {
if p.Pair.Key != "" {
return callback(&pb.RowResponse{
Headers: []*pb.ColumnInfo{
{Name: p.Field, Datatype: "string"},
{Name: "count", Datatype: "uint64"},
},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: p.Pair.Key}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: p.Pair.Count}},
},
})
} else {
return callback(&pb.RowResponse{
Headers: []*pb.ColumnInfo{
{Name: p.Field, Datatype: "uint64"},
{Name: "count", Datatype: "uint64"},
},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: p.Pair.ID}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: p.Pair.Count}},
},
})
}
}
// MarshalJSON marshals PairField into a JSON-encoded byte slice,
// excluding `Field`.
func (p PairField) MarshalJSON() ([]byte, error) {
@ -410,12 +444,60 @@ func (p Pairs) String() string {
return buf.String()
}
// PairsField
// PairsField is a Pairs object with its associated field.
type PairsField struct {
Pairs []Pair
Field string
}
// ToTable implements the ToTabler interface.
func (p *PairsField) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(p, len(p.Pairs))
}
// ToRows implements the ToRowser interface.
func (p *PairsField) ToRows(callback func(*pb.RowResponse) error) error {
// Determine if the ID has string keys.
var stringKeys bool
if len(p.Pairs) > 0 {
if p.Pairs[0].Key != "" {
stringKeys = true
}
}
dtype := "uint64"
if stringKeys {
dtype = "string"
}
ci := []*pb.ColumnInfo{
{Name: p.Field, Datatype: dtype},
{Name: "count", Datatype: "uint64"},
}
for _, pair := range p.Pairs {
if stringKeys {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: pair.Key}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.Count)}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
} else {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.ID)}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.Count)}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
}
ci = nil //only send on the first
}
return nil
}
// MarshalJSON marshals PairsField into a JSON-encoded byte slice,
// excluding `Field`.
func (p PairsField) MarshalJSON() ([]byte, error) {

185
executor.go
View file

@ -26,6 +26,7 @@ import (
"github.com/molecula/ext"
"github.com/pilosa/pilosa/v2/pql"
pb "github.com/pilosa/pilosa/v2/proto"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/shardwidth"
"github.com/pilosa/pilosa/v2/tracing"
@ -1556,6 +1557,47 @@ type RowIdentifiers struct {
field string
}
// ToTable implements the ToTabler interface.
func (r RowIdentifiers) ToTable() (*pb.TableResponse, error) {
var n int
if len(r.Keys) > 0 {
n = len(r.Keys)
} else {
n = len(r.Rows)
}
return pb.RowsToTable(&r, n)
}
// ToRows implements the ToRowser interface.
func (r RowIdentifiers) ToRows(callback func(*pb.RowResponse) error) error {
if len(r.Keys) > 0 {
ci := []*pb.ColumnInfo{{Name: r.Field(), Datatype: "string"}}
for _, key := range r.Keys {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: key}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil
}
} else {
ci := []*pb.ColumnInfo{{Name: r.Field(), Datatype: "uint64"}}
for _, id := range r.Rows {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(id)}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil
}
}
return nil
}
// Field returns the field name associated to the row.
func (r *RowIdentifiers) Field() string {
return r.field
@ -1751,6 +1793,56 @@ func (fr FieldRow) String() string {
return fmt.Sprintf("%s.%d.%s", fr.Field, fr.RowID, fr.RowKey)
}
// GroupCounts is a list of GroupCount.
type GroupCounts []GroupCount
// ToTable implements the ToTabler interface.
func (g GroupCounts) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(&g, len(g))
}
// ToRows implements the ToRowser interface.
func (g GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
for i, gc := range g {
var ci []*pb.ColumnInfo
if i == 0 {
for _, fieldRow := range gc.Group {
if fieldRow.RowKey != "" {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "string"})
} else if fieldRow.Value != nil {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "int64"})
} else {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "uint64"})
}
}
ci = append(ci, &pb.ColumnInfo{Name: "count", Datatype: "uint64"})
ci = append(ci, &pb.ColumnInfo{Name: "sum", Datatype: "int64"})
}
rowResp := &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{},
}
for _, fieldRow := range gc.Group {
if fieldRow.RowKey != "" {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: fieldRow.RowKey}})
} else if fieldRow.Value != nil {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: *fieldRow.Value}})
} else {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: fieldRow.RowID}})
}
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: gc.Count}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: gc.Sum}},
)
if err := callback(rowResp); err != nil {
return errors.Wrap(err, "calling callback")
}
}
return nil
}
// GroupCount represents a result item for a group by query.
type GroupCount struct {
Group []FieldRow `json:"group"`
@ -4127,6 +4219,46 @@ func (s *SignedRow) Field() string {
return s.field
}
// ToTable implements the ToTabler interface.
func (s SignedRow) ToTable() (*pb.TableResponse, error) {
var n uint64
if s.Neg != nil {
n += s.Neg.Count()
}
if s.Pos != nil {
n += s.Pos.Count()
}
return pb.RowsToTable(&s, int(n))
}
// ToRows implements the ToRowser interface.
func (s SignedRow) ToRows(callback func(*pb.RowResponse) error) error {
// TODO: address the overflow issue with values outside the int64 range
ci := []*pb.ColumnInfo{{Name: s.Field(), Datatype: "int64"}}
negs := s.Neg.Columns()
for i := len(negs) - 1; i >= 0; i-- {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: -1 * int64(negs[i])}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil
}
for _, id := range s.Pos.Columns() {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: int64(id)}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil
}
return nil
}
func (sr *SignedRow) union(other SignedRow) SignedRow {
ret := SignedRow{&Row{}, &Row{}, ""}
@ -4159,6 +4291,59 @@ type ValCount struct {
Count int64 `json:"count"`
}
// ToTable implements the ToTabler interface.
func (v ValCount) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(&v, 1)
}
// ToRows implements the ToRowser interface.
func (v ValCount) ToRows(callback func(*pb.RowResponse) error) error {
var ci []*pb.ColumnInfo
// ValCount can have a decimal, float, or integer value, but
// not more than one (as of this writing).
if v.DecimalVal != nil {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "decimal"},
{Name: "count", Datatype: "int64"},
}
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: v.DecimalVal.Value, Scale: v.DecimalVal.Scale}}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: v.Count}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
} else if v.FloatVal != 0 {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "float64"},
{Name: "count", Datatype: "int64"},
}
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Float64Val{Float64Val: v.FloatVal}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: v.Count}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
} else {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "int64"},
{Name: "count", Datatype: "int64"},
}
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: v.Val}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: v.Count}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
}
return nil
}
func (vc *ValCount) add(other ValCount) ValCount {
return ValCount{
Val: vc.Val + other.Val,

2
handler.go
View file

@ -57,7 +57,7 @@ type QueryRequest struct {
// QueryResponse represent a response from a processed query.
type QueryResponse struct {
// Result for each top-level query call.
// Can be a Bitmap, Pairs, or uint64.
// Can be a Bitmap, Pairs, or uint64. // TODO: this comment is out of date.
Results []interface{}
// Set of column attribute objects matching IDs returned in Result.

49
proto/interface.go
View file

@ -15,10 +15,10 @@
package pilosa
import (
"errors"
"fmt"
"strings"
"github.com/pkg/errors"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
@ -37,6 +37,53 @@ type StreamServer interface {
Send(*RowResponse) error
}
// ToTabler is an interface for any type that can
// represent itself as a TableResponse.
type ToTabler interface {
ToTable() (*TableResponse, error)
}
// ToRowser is an interface for any type that can
// represent itself as one or more RowResponses.
// ToRows takes a callback function which should be
// called for each row in the response.
type ToRowser interface {
ToRows(func(*RowResponse) error) error
}
// RowsToTable is a helper function which takes a ToRowser,
// along with the number of rows, and returns a TableResponse.
// Obviously passing the number of rows seems unnecessary,
// and we could remove that requirement, but for now we
// do it to allow for pre-allocation of the rows slice.
func RowsToTable(tr ToRowser, n int) (*TableResponse, error) {
var headers []*ColumnInfo
rows := make([]*Row, n)
// This callback gets called for every "row" in r.
// Each row populates its position in the pre-allocated
// `rows`. The headers get set based on those received
// in the first row.
var idx int
cb := func(rr *RowResponse) error {
if idx == 0 {
headers = rr.GetHeaders()
}
rows[idx] = &Row{Columns: rr.GetColumns()}
idx++
return nil
}
if err := tr.ToRows(cb); err != nil {
return nil, errors.Wrap(err, "calling callback")
}
return &TableResponse{
Headers: headers,
Rows: rows,
}, nil
}
// EOF acts as an io.EOF encoded into a RowResponse.
var EOF *RowResponse = &RowResponse{
StatusError: &StatusError{

48
row.go
View file

@ -19,6 +19,7 @@ import (
"sort"
"github.com/molecula/ext"
pb "github.com/pilosa/pilosa/v2/proto"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pkg/errors"
)
@ -84,6 +85,53 @@ func NewRowFromRoaring(data []byte) *Row {
return r
}
// ToTable implements the ToTabler interface.
func (r *Row) ToTable() (*pb.TableResponse, error) {
var n int
if len(r.Keys) > 0 {
n = len(r.Keys)
} else {
n = len(r.Columns())
}
return pb.RowsToTable(r, n)
}
// ToRows implements the ToRowser interface.
func (r *Row) ToRows(callback func(*pb.RowResponse) error) error {
if len(r.Keys) > 0 {
// Column keys
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "string"},
}
for _, x := range r.Keys {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: x}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil //only send on the first
}
} else {
// Column IDs
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "uint64"},
}
for _, x := range r.Columns() {
if err := callback(&pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: x}},
}}); err != nil {
return errors.Wrap(err, "calling callback")
}
ci = nil //only send on the first
}
}
return nil
}
// Roaring returns the row treated as a unified roaring bitmap.
func (r *Row) Roaring() []byte {
bitmaps := make([]*roaring.Bitmap, len(r.segments))

423
server/grpc.go
View file

@ -44,9 +44,12 @@ type grpcHandler struct {
}
// errorToStatusError appends an appropriate grpc status code
// to the error (returning it as a status.Error). It is
// assumed that the input err is non-nil.
// to the error (returning it as a status.Error).
func errToStatusError(err error) error {
if err == nil {
return status.New(codes.OK, "").Err()
}
// Check error string.
switch errors.Cause(err) {
case pilosa.ErrIndexNotFound, pilosa.ErrFieldNotFound:
@ -68,28 +71,36 @@ func (h grpcHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQL
}
t := time.Now()
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
durQuery := time.Since(t)
// TODO: what about resp.Err?
// TODO: what about resp.CollumnAttrSets?
if err != nil {
return errToStatusError(err)
} else if len(resp.Results) != 1 {
// TODO: make a test for this
return status.Error(codes.InvalidArgument, "QueryPQL handles exactly one query")
}
longQueryTime := h.api.LongQueryTime()
if longQueryTime > 0 && durQuery > longQueryTime {
h.logger.Printf("GRPC QueryPQL %v %s", durQuery, query.Query)
}
rslt := resp.Results[0]
toRowser, err := toRowserWrapper(rslt)
if err != nil {
return errors.Wrap(err, "wrapping as type ToRowser")
}
t = time.Now()
for row := range makeRows(resp, h.logger) {
err = stream.Send(row)
if err != nil {
return errToStatusError(err)
}
if err := toRowser.ToRows(stream.Send); err != nil {
return errToStatusError(err)
}
durFormat := time.Since(t)
h.stats.Timing(pilosa.MetricGRPCStreamQueryDurationSeconds, durQuery, 0.1)
h.stats.Timing(pilosa.MetricGRPCStreamFormatDurationSeconds, durFormat, 0.1)
return nil
return errToStatusError(nil)
}
// QueryPQLUnary is a unary-response (non-streaming) version of QueryPQL, returning a TableResponse.
@ -100,31 +111,116 @@ func (h grpcHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest)
}
t := time.Now()
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(ctx, &query)
durQuery := time.Since(t)
if err != nil {
return nil, errToStatusError(err)
} else if len(resp.Results) != 1 {
// TODO: make a test for this
return nil, status.Error(codes.InvalidArgument, "QueryPQLUnary handles exactly one query")
}
longQueryTime := h.api.LongQueryTime()
if longQueryTime > 0 && durQuery > longQueryTime {
h.logger.Printf("GRPC QueryPQLUnary %v %s", durQuery, query.Query)
}
t = time.Now()
response := &pb.TableResponse{
Rows: make([]*pb.Row, 0),
rslt := resp.Results[0]
toTabler, err := toTablerWrapper(rslt)
if err != nil {
return nil, errors.Wrap(err, "wrapping as type ToTabler")
}
for row := range makeRows(resp, h.logger) {
if len(row.Headers) != 0 {
response.Headers = row.Headers
}
response.Rows = append(response.Rows, &pb.Row{Columns: row.Columns})
t = time.Now()
table, err := toTabler.ToTable()
if err != nil {
return nil, errToStatusError(err)
}
durFormat := time.Since(t)
h.stats.Timing(pilosa.MetricGRPCUnaryQueryDurationSeconds, durQuery, 0.1)
h.stats.Timing(pilosa.MetricGRPCUnaryFormatDurationSeconds, durFormat, 0.1)
return response, nil
return table, errToStatusError(nil)
}
// ResultUint64 is a wrapper around a uint64 result type
// so that we can implement the ToTabler and ToRowser
// interfaces.
type ResultUint64 uint64
// ToTable implements the ToTabler interface.
func (r ResultUint64) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(&r, 1)
}
// ToRows implements the ToRowser interface.
func (r ResultUint64) ToRows(callback func(*pb.RowResponse) error) error {
return callback(&pb.RowResponse{
Headers: []*pb.ColumnInfo{{Name: "count", Datatype: "uint64"}},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(r)}},
}})
}
// ResultBool is a wrapper around a bool result type
// so that we can implement the ToTabler and ToRowser
// interfaces.
type ResultBool bool
// ToTable implements the ToTabler interface.
func (r ResultBool) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(&r, 1)
}
// ToRows implements the ToRowser interface.
func (r ResultBool) ToRows(callback func(*pb.RowResponse) error) error {
return callback(&pb.RowResponse{
Headers: []*pb.ColumnInfo{{Name: "result", Datatype: "bool"}},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bool(r)}},
}})
}
// Normally we wouldn't need this wrapper, but since pilosa returns
// some concrete types for which we can't implement the ToTabler
// interface, we have to check for those here and then wrap them
// with a custom type.
func toTablerWrapper(result interface{}) (pb.ToTabler, error) {
toTabler, ok := result.(pb.ToTabler)
if !ok {
switch v := result.(type) {
case []pilosa.GroupCount:
toTabler = pilosa.GroupCounts(v)
case uint64:
toTabler = ResultUint64(v)
case bool:
toTabler = ResultBool(v)
default:
return nil, errors.Errorf("ToTabler interface not implemented by type: %T", result)
}
}
return toTabler, nil
}
// Normally we wouldn't need this wrapper, but since pilosa returns
// some concrete types for which we can't implement the ToRowser
// interface, we have to check for those here and then wrap them
// with a custom type.
func toRowserWrapper(result interface{}) (pb.ToRowser, error) {
toRowser, ok := result.(pb.ToRowser)
if !ok {
switch v := result.(type) {
case []pilosa.GroupCount:
toRowser = pilosa.GroupCounts(v)
case uint64:
toRowser = ResultUint64(v)
case bool:
toRowser = ResultBool(v)
default:
return nil, errors.Errorf("ToRowser interface not implemented by type: %T", result)
}
}
return toRowser, nil
}
// fieldDataType returns a useful data type (string,
@ -161,7 +257,7 @@ func fieldDataType(f *pilosa.Field) string {
func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectServer) error {
const defaultLimit = 100000
index, err := h.api.Index(context.Background(), req.Index)
index, err := h.api.Index(stream.Context(), req.Index)
if err != nil {
return errToStatusError(err)
}
@ -223,7 +319,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrapf(err, "querying for all: %s", pql)
}
@ -260,7 +356,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrapf(err, "querying rows for set: %s", pql)
}
@ -284,7 +380,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "querying rows for mutex")
}
@ -316,7 +412,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
if err != nil {
return errors.Wrap(err, "getting int value")
} else if ok {
vals, err := h.api.TranslateIndexIDs(context.Background(), fi, []uint64{uint64(intVal)})
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(intVal)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
@ -369,7 +465,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "querying rows for bool")
}
@ -442,7 +538,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrapf(err, "querying for all: %s", pql)
}
@ -479,7 +575,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "querying set rows(keys)")
}
@ -503,7 +599,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "querying mutex rows(keys)")
}
@ -526,7 +622,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
case "int":
// Translate column key.
id, err := h.api.TranslateIndexKey(context.Background(), index.Name(), col)
id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col)
if err != nil {
return errors.Wrap(err, "translating column key")
}
@ -541,7 +637,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
if err != nil {
return errors.Wrap(err, "getting int value")
} else if ok {
vals, err := h.api.TranslateIndexIDs(context.Background(), fi, []uint64{uint64(intVal)})
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(intVal)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
@ -578,7 +674,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
case "decimal":
// Translate column key.
id, err := h.api.TranslateIndexKey(context.Background(), index.Name(), col)
id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col)
if err != nil {
return errors.Wrap(err, "translating column key")
}
@ -600,7 +696,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
Index: req.Index,
Query: pql,
}
resp, err := h.api.Query(context.Background(), &query)
resp, err := h.api.Query(stream.Context(), &query)
if err != nil {
return errors.Wrap(err, "querying bool rows(keys)")
}
@ -637,269 +733,6 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
return nil
}
// I think ideally this would be plugged in the executor somewhere
// in order to get some concurrency benefit but we can
// start with the combined response
func makeRows(resp pilosa.QueryResponse, logger logger.Logger) chan *pb.RowResponse {
results := make(chan *pb.RowResponse)
go func() {
var breakLoop bool // Support the "break" inside the switch.
for _, result := range resp.Results {
if breakLoop {
break
}
switch r := result.(type) {
case *pilosa.Row:
if len(r.Keys) > 0 {
// Column keys
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "string"},
}
for _, x := range r.Keys {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: x}},
}}
ci = nil //only send on the first
}
} else {
// Column IDs
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "uint64"},
}
for _, x := range r.Columns() {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: x}},
}}
ci = nil //only send on the first
}
// The following will return roaring segments.
// This is commented out for now until we decide how we want to use this.
/*
// Roaring segments
ci := []*pb.ColumnInfo{
// TODO:
{Name: "shard", Datatype: "uint64"},
{Name: "segment", Datatype: "roaring"},
}
for _, x := range r.Segments() {
shard, b := x.Raw()
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_IntVal{int64(shard)}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BlobVal{b}},
}}
ci = nil //only send on the first
}
*/
}
case pilosa.PairField:
if r.Pair.Key != "" {
results <- &pb.RowResponse{
Headers: []*pb.ColumnInfo{
{Name: r.Field, Datatype: "string"},
{Name: "count", Datatype: "uint64"},
},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: r.Pair.Key}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: r.Pair.Count}},
},
}
} else {
results <- &pb.RowResponse{
Headers: []*pb.ColumnInfo{
{Name: r.Field, Datatype: "uint64"},
{Name: "count", Datatype: "uint64"},
},
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: r.Pair.ID}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: r.Pair.Count}},
},
}
}
case *pilosa.PairsField:
// Determine if the ID has string keys.
var stringKeys bool
if len(r.Pairs) > 0 {
if r.Pairs[0].Key != "" {
stringKeys = true
}
}
dtype := "uint64"
if stringKeys {
dtype = "string"
}
ci := []*pb.ColumnInfo{
{Name: r.Field, Datatype: dtype},
{Name: "count", Datatype: "uint64"},
}
for _, pair := range r.Pairs {
if stringKeys {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: pair.Key}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.Count)}},
},
}
} else {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.ID)}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(pair.Count)}},
},
}
}
ci = nil //only send on the first
}
case []pilosa.GroupCount:
for i, gc := range r {
var ci []*pb.ColumnInfo
if i == 0 {
for _, fieldRow := range gc.Group {
if fieldRow.RowKey != "" {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "string"})
} else if fieldRow.Value != nil {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "int64"})
} else {
ci = append(ci, &pb.ColumnInfo{Name: fieldRow.Field, Datatype: "uint64"})
}
}
ci = append(ci, &pb.ColumnInfo{Name: "count", Datatype: "uint64"})
ci = append(ci, &pb.ColumnInfo{Name: "sum", Datatype: "int64"})
}
rowResp := &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{},
}
for _, fieldRow := range gc.Group {
if fieldRow.RowKey != "" {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: fieldRow.RowKey}})
} else if fieldRow.Value != nil {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: *fieldRow.Value}})
} else {
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: fieldRow.RowID}})
}
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: gc.Count}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: gc.Sum}},
)
results <- rowResp
}
case pilosa.RowIdentifiers:
if len(r.Keys) > 0 {
ci := []*pb.ColumnInfo{{Name: r.Field(), Datatype: "string"}}
for _, key := range r.Keys {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: key}},
}}
ci = nil
}
} else {
ci := []*pb.ColumnInfo{{Name: r.Field(), Datatype: "uint64"}}
for _, id := range r.Rows {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(id)}},
}}
ci = nil
}
}
case uint64:
ci := []*pb.ColumnInfo{{Name: "count", Datatype: "uint64"}}
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(r)}},
}}
case bool:
ci := []*pb.ColumnInfo{{Name: "result", Datatype: "bool"}}
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: r}},
}}
case pilosa.ValCount:
var ci []*pb.ColumnInfo
// ValCount can have a decimal, float, or integer value, but
// not more than one (as of this writing).
if r.DecimalVal != nil {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "decimal"},
{Name: "count", Datatype: "int64"},
}
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: r.DecimalVal.Value, Scale: r.DecimalVal.Scale}}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: r.Count}},
}}
} else if r.FloatVal != 0 {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "float64"},
{Name: "count", Datatype: "int64"},
}
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Float64Val{Float64Val: r.FloatVal}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: r.Count}},
}}
} else {
ci = []*pb.ColumnInfo{
{Name: "value", Datatype: "int64"},
{Name: "count", Datatype: "int64"},
}
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: r.Val}},
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: r.Count}},
}}
}
case pilosa.SignedRow:
// TODO: address the overflow issue with values outside the int64 range
ci := []*pb.ColumnInfo{{Name: r.Field(), Datatype: "int64"}}
negs := r.Neg.Columns()
for i := len(negs) - 1; i >= 0; i-- {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: -1 * int64(negs[i])}},
}}
ci = nil
}
for _, id := range r.Pos.Columns() {
results <- &pb.RowResponse{
Headers: ci,
Columns: []*pb.ColumnResponse{
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: int64(id)}},
}}
ci = nil
}
default:
logger.Printf("unhandled %T\n", r)
breakLoop = true
}
}
close(results)
}()
return results
}
type grpcServer struct {
api *pilosa.API
grpcServer *grpc.Server

129
server/grpc_internal_test.go
View file

@ -18,11 +18,10 @@ import (
"testing"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/logger"
)
func TestGRPC(t *testing.T) {
t.Run("makeRows", func(t *testing.T) {
t.Run("ToTable", func(t *testing.T) {
type expHeader struct {
name string
dataType string
@ -36,16 +35,6 @@ func TestGRPC(t *testing.T) {
expHeaders []expHeader
expColumns [][]expColumn
}{
{
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)}},
},
// Row (uint64)
{
pilosa.NewRow(10, 11, 12),
@ -70,11 +59,14 @@ func TestGRPC(t *testing.T) {
{"twelve"},
},
},
// Pair (uint64)
// PairField (uint64)
{
pilosa.Pair{ID: 10, Count: 123},
pilosa.PairField{
Pair: pilosa.Pair{ID: 10, Count: 123},
Field: "fld",
},
[]expHeader{
{"_id", "uint64"},
{"fld", "uint64"},
{"count", "uint64"},
},
[][]expColumn{
@ -83,23 +75,29 @@ func TestGRPC(t *testing.T) {
},
// Pair (string)
{
pilosa.Pair{Key: "ten", Count: 123},
pilosa.PairField{
Pair: pilosa.Pair{Key: "ten", Count: 123},
Field: "fld",
},
[]expHeader{
{"_id", "string"},
{"fld", "string"},
{"count", "uint64"},
},
[][]expColumn{
{string("ten"), uint64(123)},
},
},
// []Pair (uint64)
// *PairsField (uint64)
{
[]pilosa.Pair{
{ID: 10, Count: 123},
{ID: 11, Count: 456},
&pilosa.PairsField{
Pairs: []pilosa.Pair{
{ID: 10, Count: 123},
{ID: 11, Count: 456},
},
Field: "fld",
},
[]expHeader{
{"_id", "uint64"},
{"fld", "uint64"},
{"count", "uint64"},
},
[][]expColumn{
@ -107,14 +105,17 @@ func TestGRPC(t *testing.T) {
{uint64(11), uint64(456)},
},
},
// []Pair (string)
// *PairsField (string)
{
[]pilosa.Pair{
{Key: "ten", Count: 123},
{Key: "eleven", Count: 456},
&pilosa.PairsField{
Pairs: []pilosa.Pair{
{Key: "ten", Count: 123},
{Key: "eleven", Count: 456},
},
Field: "fld",
},
[]expHeader{
{"_id", "string"},
{"fld", "string"},
{"count", "uint64"},
},
[][]expColumn{
@ -236,35 +237,61 @@ func TestGRPC(t *testing.T) {
{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)},
},
},
}
logger := logger.NopLogger
for ti, test := range tests {
results := make([]interface{}, 0)
results = append(results, test.result)
toTabler, err := toTablerWrapper(test.result)
if err != nil {
t.Fatal(err)
}
table, err := toTabler.ToTable()
if err != nil {
t.Fatal(err)
}
qr := pilosa.QueryResponse{}
qr.Results = results
ch := makeRows(qr, logger)
cnt := 0
for row := range ch {
// Ensure headers match (on the first row).
if cnt == 0 {
for i, header := range row.GetHeaders() {
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 headers match.
for i, header := range table.GetHeaders() {
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, column := range row.GetColumns() {
switch v := test.expColumns[cnt][i].(type) {
// Ensure column data matches.
for i, row := range table.GetRows() {
for j, column := range row.GetColumns() {
switch v := test.expColumns[i][j].(type) {
case string:
val := column.GetStringVal()
if val != v {
@ -294,8 +321,6 @@ func TestGRPC(t *testing.T) {
t.Fatalf("test %d has unhandled data type: %T", ti, v)
}
}
cnt++
}
}
})