mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-06 16:45:55 +00:00
7321f9427c
this involved adding an optional float value to the ValCount struct which complicated result types, necessitated grpc changes, and needed quite a few tests at different layers.
908 lines
26 KiB
Go
908 lines
26 KiB
Go
// Copyright 2017 Pilosa Corp.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package server
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"fmt"
|
|
"net"
|
|
"strings"
|
|
|
|
"github.com/pilosa/pilosa/v2"
|
|
"github.com/pilosa/pilosa/v2/logger"
|
|
pb "github.com/pilosa/pilosa/v2/proto"
|
|
"github.com/pkg/errors"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/reflection"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
// grpcHandler contains methods which handle the various gRPC requests.
|
|
type grpcHandler struct {
|
|
api *pilosa.API
|
|
|
|
logger logger.Logger
|
|
}
|
|
|
|
// 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.
|
|
func errToStatusError(err error) error {
|
|
// Check error string.
|
|
switch errors.Cause(err) {
|
|
case pilosa.ErrIndexNotFound, pilosa.ErrFieldNotFound:
|
|
return status.Error(codes.NotFound, err.Error())
|
|
}
|
|
// Check error type.
|
|
switch errors.Cause(err).(type) {
|
|
case pilosa.NotFoundError:
|
|
return status.Error(codes.NotFound, err.Error())
|
|
}
|
|
return status.Error(codes.Unknown, err.Error())
|
|
}
|
|
|
|
// QueryPQL handles the PQL request and sends RowResponses to the stream.
|
|
func (h grpcHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQLServer) error {
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: req.Pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errToStatusError(err)
|
|
}
|
|
for row := range makeRows(resp, h.logger) {
|
|
err = stream.Send(row)
|
|
if err != nil {
|
|
return errToStatusError(err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// fieldDataType returns a useful data type (string,
|
|
// uint64, bool, etc.) based on the Pilosa field type.
|
|
func fieldDataType(f *pilosa.Field) string {
|
|
switch f.Type() {
|
|
case "set":
|
|
if f.Keys() {
|
|
return "[]string"
|
|
}
|
|
return "[]uint64"
|
|
case "mutex":
|
|
if f.Keys() {
|
|
return "string"
|
|
}
|
|
return "uint64"
|
|
case "int":
|
|
if f.Keys() {
|
|
return "string"
|
|
}
|
|
return "int64"
|
|
case "decimal":
|
|
return "float64"
|
|
case "bool":
|
|
return "bool"
|
|
case "time":
|
|
return "int64" // TODO: this is a placeholder
|
|
default:
|
|
panic(fmt.Sprintf("unimplemented fieldDataType: %s", f.Type()))
|
|
}
|
|
}
|
|
|
|
// Inspect handles the inspect request and sends an InspectResponse to the stream.
|
|
func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectServer) error {
|
|
const defaultLimit = 100000
|
|
|
|
index, err := h.api.Index(context.Background(), req.Index)
|
|
if err != nil {
|
|
return errToStatusError(err)
|
|
}
|
|
|
|
var fields []*pilosa.Field
|
|
for _, field := range index.Fields() {
|
|
// exclude internal fields (starting with "_")
|
|
if strings.HasPrefix(field.Name(), "_") {
|
|
continue
|
|
}
|
|
if len(req.FilterFields) > 0 {
|
|
for _, filter := range req.FilterFields {
|
|
if filter == field.Name() {
|
|
fields = append(fields, field)
|
|
break
|
|
}
|
|
|
|
}
|
|
} else {
|
|
fields = append(fields, field)
|
|
}
|
|
}
|
|
|
|
limit := req.Limit
|
|
if limit == 0 {
|
|
limit = defaultLimit
|
|
}
|
|
offset := req.Offset
|
|
|
|
if !index.Keys() {
|
|
ints, ok := req.Columns.Type.(*pb.IdsOrKeys_Ids)
|
|
if !ok {
|
|
return errors.New("invalid int columns")
|
|
}
|
|
ci := []*pb.ColumnInfo{
|
|
{Name: "_id", Datatype: "uint64"},
|
|
}
|
|
for _, field := range fields {
|
|
ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fieldDataType(field)})
|
|
}
|
|
|
|
// If Columns is empty, then get the _exists list (via All()),
|
|
// from the index and loop over that instead.
|
|
cols := ints.Ids.Vals
|
|
if len(cols) > 0 {
|
|
// Apply limit/offset to the provided columns.
|
|
if int(offset) >= len(cols) {
|
|
return nil
|
|
}
|
|
end := limit + offset
|
|
if int(end) > len(cols) {
|
|
end = uint64(len(cols))
|
|
}
|
|
cols = cols[offset:end]
|
|
} else {
|
|
// Prevent getting too many records by forcing a limit.
|
|
pql := fmt.Sprintf("All(limit=%d, offset=%d)", limit, offset)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "querying for all: %s", pql)
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(*pilosa.Row)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting results as a row")
|
|
}
|
|
|
|
limitedCols := ids.Columns()
|
|
if len(limitedCols) == 0 {
|
|
// If cols is still empty after the limit/offset, then
|
|
// return with no results.
|
|
return nil
|
|
}
|
|
cols = limitedCols
|
|
}
|
|
|
|
for _, col := range cols {
|
|
rowResp := &pb.RowResponse{
|
|
Headers: ci,
|
|
Columns: []*pb.ColumnResponse{
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: col}},
|
|
},
|
|
}
|
|
ci = nil // only include headers with the first row
|
|
|
|
for _, field := range fields {
|
|
// TODO: handle `time` fields
|
|
switch field.Type() {
|
|
case "set":
|
|
pql := fmt.Sprintf("Rows(%s, column=%d)", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "querying rows for set: %s", pql)
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Keys) > 0 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
|
|
}
|
|
|
|
case "mutex":
|
|
pql := fmt.Sprintf("Rows(%s, column=%d)", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrap(err, "querying rows for mutex")
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Keys) == 1 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
|
|
} else if len(ids.Rows) == 1 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "int":
|
|
if field.Keys() {
|
|
var value string
|
|
var exists bool
|
|
var err error
|
|
if fi := field.ForeignIndex(); fi != "" {
|
|
// Get the value from the int field.
|
|
intVal, ok, err := field.Value(col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting int value")
|
|
} else if ok {
|
|
vals, err := h.api.TranslateIndexIDs(context.Background(), fi, []uint64{uint64(intVal)})
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting keys for ids")
|
|
}
|
|
if len(vals) > 0 && vals[0] != "" {
|
|
value = vals[0]
|
|
exists = true
|
|
}
|
|
}
|
|
} else {
|
|
value, exists, err = field.StringValue(col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting string field value for column")
|
|
}
|
|
}
|
|
if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
} else {
|
|
value, exists, err := field.Value(col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting int field value for column")
|
|
} else if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
}
|
|
|
|
case "decimal":
|
|
value, exists, err := field.FloatValue(col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting decimal field value for column")
|
|
} else if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Float64Val{Float64Val: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "bool":
|
|
pql := fmt.Sprintf("Rows(%s, column=%d)", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrap(err, "querying rows for bool")
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Rows) == 1 {
|
|
var bval bool
|
|
if ids.Rows[0] == 1 {
|
|
bval = true
|
|
}
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "time":
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
}
|
|
|
|
if err := stream.Send(rowResp); err != nil {
|
|
return errors.Wrap(err, "sending response to stream")
|
|
}
|
|
}
|
|
|
|
} else {
|
|
var cols []string
|
|
|
|
switch keys := req.Columns.Type.(type) {
|
|
case *pb.IdsOrKeys_Ids:
|
|
// The default behavior (in api/client/grpc.go) is to
|
|
// send an empty set of Ids even if the index supports
|
|
// keys, so in that case we just need to ignore it.
|
|
case *pb.IdsOrKeys_Keys:
|
|
cols = keys.Keys.Vals
|
|
default:
|
|
return errToStatusError(errors.New("invalid key columns"))
|
|
}
|
|
|
|
ci := []*pb.ColumnInfo{
|
|
{Name: "_id", Datatype: "string"},
|
|
}
|
|
for _, field := range fields {
|
|
ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: fieldDataType(field)})
|
|
}
|
|
|
|
// If Columns is empty, then get the _exists list (via All()),
|
|
// from the index and loop over that instead.
|
|
if len(cols) > 0 {
|
|
// Apply limit/offset to the provided columns.
|
|
if int(offset) >= len(cols) {
|
|
return nil
|
|
}
|
|
end := limit + offset
|
|
if int(end) > len(cols) {
|
|
end = uint64(len(cols))
|
|
}
|
|
cols = cols[offset:end]
|
|
} else {
|
|
// Prevent getting too many records by forcing a limit.
|
|
pql := fmt.Sprintf("All(limit=%d, offset=%d)", limit, offset)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "querying for all: %s", pql)
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(*pilosa.Row)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting results as a row")
|
|
}
|
|
|
|
limitedCols := ids.Keys
|
|
if len(limitedCols) == 0 {
|
|
// If cols is still empty after the limit/offset, then
|
|
// return with no results.
|
|
return nil
|
|
}
|
|
cols = limitedCols
|
|
}
|
|
|
|
for _, col := range cols {
|
|
rowResp := &pb.RowResponse{
|
|
Headers: ci,
|
|
Columns: []*pb.ColumnResponse{
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: col}},
|
|
},
|
|
}
|
|
ci = nil // only include headers with the first row
|
|
|
|
for _, field := range fields {
|
|
// TODO: handle `time` fields
|
|
switch field.Type() {
|
|
case "set":
|
|
pql := fmt.Sprintf("Rows(%s, column=\"%s\")", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrap(err, "querying set rows(keys)")
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Keys) > 0 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
|
|
}
|
|
|
|
case "mutex":
|
|
pql := fmt.Sprintf("Rows(%s, column=\"%s\")", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrap(err, "querying mutex rows(keys)")
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Keys) == 1 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
|
|
} else if len(ids.Rows) == 1 {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "int":
|
|
// Translate column key.
|
|
id, err := h.api.TranslateIndexKey(context.Background(), index.Name(), col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "translating column key")
|
|
}
|
|
|
|
if field.Keys() {
|
|
var value string
|
|
var exists bool
|
|
var err error
|
|
if fi := field.ForeignIndex(); fi != "" {
|
|
// Get the value from the int field.
|
|
intVal, ok, err := field.Value(id)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting int value")
|
|
} else if ok {
|
|
vals, err := h.api.TranslateIndexIDs(context.Background(), fi, []uint64{uint64(intVal)})
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting keys for ids")
|
|
}
|
|
if len(vals) > 0 && vals[0] != "" {
|
|
value = vals[0]
|
|
exists = true
|
|
}
|
|
}
|
|
} else {
|
|
value, exists, err = field.StringValue(id)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting string field value for column")
|
|
}
|
|
}
|
|
if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
} else {
|
|
value, exists, err := field.Value(id)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting int field value for column")
|
|
} else if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
}
|
|
|
|
case "decimal":
|
|
// Translate column key.
|
|
id, err := h.api.TranslateIndexKey(context.Background(), index.Name(), col)
|
|
if err != nil {
|
|
return errors.Wrap(err, "translating column key")
|
|
}
|
|
|
|
value, exists, err := field.FloatValue(id)
|
|
if err != nil {
|
|
return errors.Wrap(err, "getting decimal field value for column")
|
|
} else if exists {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Float64Val{Float64Val: value}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "bool":
|
|
pql := fmt.Sprintf("Rows(%s, column=\"%s\")", field.Name(), col)
|
|
query := pilosa.QueryRequest{
|
|
Index: req.Index,
|
|
Query: pql,
|
|
}
|
|
resp, err := h.api.Query(context.Background(), &query)
|
|
if err != nil {
|
|
return errors.Wrap(err, "querying bool rows(keys)")
|
|
}
|
|
|
|
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
|
|
if !ok {
|
|
return errors.Wrap(err, "getting row identifiers")
|
|
}
|
|
|
|
if len(ids.Rows) == 1 {
|
|
var bval bool
|
|
if ids.Rows[0] == 1 {
|
|
bval = true
|
|
}
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
|
|
} else {
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
|
|
case "time":
|
|
rowResp.Columns = append(rowResp.Columns,
|
|
&pb.ColumnResponse{ColumnVal: nil})
|
|
}
|
|
}
|
|
|
|
if err := stream.Send(rowResp); err != nil {
|
|
return errors.Wrap(err, "sending response to stream")
|
|
}
|
|
}
|
|
|
|
}
|
|
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 {
|
|
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 {
|
|
rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: uint64(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 float or integeger value, but
|
|
// not both (as of this writing).
|
|
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
|
|
hostPort string
|
|
|
|
logger logger.Logger
|
|
}
|
|
|
|
type grpcServerOption func(s *grpcServer) error
|
|
|
|
func OptGRPCServerAPI(api *pilosa.API) grpcServerOption {
|
|
return func(s *grpcServer) error {
|
|
s.api = api
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func OptGRPCServerURI(uri *pilosa.URI) grpcServerOption {
|
|
hostport := fmt.Sprintf("%s:%d", uri.Host, uri.Port)
|
|
return func(s *grpcServer) error {
|
|
s.hostPort = hostport
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func OptGRPCServerLogger(logger logger.Logger) grpcServerOption {
|
|
return func(s *grpcServer) error {
|
|
s.logger = logger
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (s *grpcServer) Serve(tlsConfig *tls.Config) error {
|
|
// create listener
|
|
lis, err := net.Listen("tcp", s.hostPort)
|
|
if err != nil {
|
|
return errors.Wrap(err, "creating listener")
|
|
}
|
|
s.logger.Printf("enabled grpc listening on %s", lis.Addr())
|
|
|
|
opts := make([]grpc.ServerOption, 0)
|
|
if tlsConfig != nil {
|
|
creds := credentials.NewTLS(tlsConfig)
|
|
opts = append(opts, grpc.Creds(creds))
|
|
}
|
|
|
|
// create grpc server
|
|
s.grpcServer = grpc.NewServer(opts...)
|
|
pb.RegisterPilosaServer(s.grpcServer, grpcHandler{api: s.api, logger: s.logger})
|
|
|
|
// register the server so its services are available to grpc_cli and others
|
|
reflection.Register(s.grpcServer)
|
|
|
|
// and start...
|
|
if err := s.grpcServer.Serve(lis); err != nil {
|
|
return errors.Wrap(err, "starting grpc server")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func NewGRPCServer(opts ...grpcServerOption) (*grpcServer, error) {
|
|
server := &grpcServer{
|
|
logger: logger.NopLogger,
|
|
}
|
|
for _, opt := range opts {
|
|
err := opt(server)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "applying option")
|
|
}
|
|
}
|
|
return server, nil
|
|
}
|