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

Merge pull request #50 from travisturner/datatype-fixes

Browse source 
Add StreamClient and StreamServer interfaces
This commit is contained in:
Travis Turner 2019-11-27 17:21:55 -06:00 • committed by GitHub
commit bc0018b67b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 40 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
api/client/grpc.go
View file

@ -58,7 +58,7 @@ func (c *GRPCClient) Close() error {
}
// Query returns a stream of RowResponse for the given index and PQL string.
func (c *GRPCClient) Query(ctx context.Context, index string, pql string) (grpc.ClientStream, error) {
func (c *GRPCClient) Query(ctx context.Context, index string, pql string) (pb.StreamClient, error) {
if c.conn == nil {
return nil, errors.New("client has not established a grpc connection")
}
@ -79,8 +79,8 @@ func (c *GRPCClient) Query(ctx context.Context, index string, pql string) (grpc.
}
// Inspect returns a stream of RowResponse for the given index, columns, and filters.
// It is inteded to mimic something like "select [fields] from table where recordID IN (...)".
func (c *GRPCClient) Inspect(ctx context.Context, index string, columnIDs []uint64, columnKeys []string, fieldFilters []string) (grpc.ClientStream, error) {
// It is intended to mimic something like "select [fields] from table where recordID IN (...)".
func (c *GRPCClient) Inspect(ctx context.Context, index string, columnIDs []uint64, columnKeys []string, fieldFilters []string) (pb.StreamClient, error) {
if c.conn == nil {
return nil, errors.New("client has not established a grpc connection")
}

23
proto/interface.go Normal file
View file

@ -0,0 +1,23 @@
// 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 pilosa
type StreamClient interface {
Recv() (*RowResponse, error)
}
type StreamServer interface {
Send(*RowResponse) error
}

14
server/grpc.go
View file

@ -205,6 +205,9 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "time":
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
@ -548,6 +551,17 @@ func makeRows(resp pilosa.QueryResponse, logger logger.Logger) chan *pb.RowRespo
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}},
}}
default:
logger.Printf("unhandled %T\n", r)
breakLoop = true