mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
e40400b130
For Fields with ForeignIndex (which have keys), the API was missing the logic to do that translation against the translateStore of the foreign index. This commit adds that logic, as well as some missing translateStore-related logic in the gRPC code.
888 lines
26 KiB
Go
888 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})
|
|
}
|
|
}
|
|
}
|
|
|
|
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:
|
|
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
|
|
}
|