diff --git a/cache.go b/cache.go index 99a13640e..1c00ed6fd 100644 --- a/cache.go +++ b/cache.go @@ -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) { diff --git a/executor.go b/executor.go index a78735755..d9622329d 100644 --- a/executor.go +++ b/executor.go @@ -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, diff --git a/handler.go b/handler.go index b726c6dff..486b99217 100644 --- a/handler.go +++ b/handler.go @@ -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. diff --git a/proto/interface.go b/proto/interface.go index 61a5dcd49..34bf53829 100644 --- a/proto/interface.go +++ b/proto/interface.go @@ -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{ diff --git a/row.go b/row.go index 7c04f60e4..e50c6ab67 100644 --- a/row.go +++ b/row.go @@ -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)) diff --git a/server/grpc.go b/server/grpc.go index dcac5cc02..e0edc77ac 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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 diff --git a/server/grpc_internal_test.go b/server/grpc_internal_test.go index 54d57468a..0fc4fd7dd 100644 --- a/server/grpc_internal_test.go +++ b/server/grpc_internal_test.go @@ -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++ } } })