scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'enterprise' into unionAA

Browse source
This commit is contained in:
seebs 2019-12-20 13:28:57 -06:00 • committed by GitHub
commit d4d3d75e28
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 760 additions and 511 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Dockerfile-clustertests
View file

@ -8,10 +8,7 @@ LABEL maintainer "dev@pilosa.com"
COPY . /go/src/github.com/pilosa/pilosa/
RUN cd /go/src/github.com/pilosa/pilosa \
&& GO111MODULE=on make vendor
RUN cd /go/src/github.com/pilosa/pilosa \
&& CGO_ENABLED=0 make install FLAGS="-a"
&& CGO_ENABLED=0 make install FLAGS="-a -mod=vendor"
# download pumba for fault injection
ADD https://github.com/alexei-led/pumba/releases/download/0.6.0/pumba_linux_amd64 /pumba

8
Makefile
View file

@ -16,12 +16,16 @@ RELEASE_ENABLED = $(subst 0,,$(RELEASE))
BUILD_TAGS += $(if $(ENTERPRISE_ENABLED),enterprise)
BUILD_TAGS += $(if $(RELEASE_ENABLED),release)
BUILD_TAGS += shardwidth$(SHARD_WIDTH)
BUILD_TAGS += $(foreach p,$(PLUGINS),plugin$(p))
define LICENSE_HASH_CODE
head -13 $1 | sed -e 's/Copyright 20[0-9][0-9]/Copyright 20XX/g' | shasum | cut -f 1 -d " "
endef
LICENSE_HASH=$(shell $(call LICENSE_HASH_CODE, pilosa.go))
PLUGINS=distinct
export GO111MODULE=on
export GOPRIVATE=github.com/molecula
export PLUGINS
# Run tests and compile Pilosa
default: test build
@ -85,14 +89,14 @@ DOCKER_COMPOSE=internal/clustertests/docker-compose.yml
# running. This will catch changes to internal/clustertests/*.go, but if you
# make changes to Pilosa, you'll want to run clustertests-build to rebuild the
# pilosa image.
clustertests:
clustertests: vendor
docker-compose -f $(DOCKER_COMPOSE) down
docker-compose -f $(DOCKER_COMPOSE) build client1
docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1
# Like clustertests, but rebuilds all images.
clustertests-build:
clustertests-build: vendor
docker-compose -f $(DOCKER_COMPOSE) down
docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build

5
api/client/grpc.go
View file

@ -80,7 +80,7 @@ func (c *GRPCClient) Query(ctx context.Context, index string, pql string) (pb.St
// Inspect returns a stream of RowResponse for the given index, columns, and filters.
// 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) {
func (c *GRPCClient) Inspect(ctx context.Context, index string, columnIDs []uint64, columnKeys []string, fieldFilters []string, limit, offset uint64) (pb.StreamClient, error) {
if c.conn == nil {
return nil, errors.New("client has not established a grpc connection")
}
@ -103,7 +103,10 @@ func (c *GRPCClient) Inspect(ctx context.Context, index string, columnIDs []uint
Index: index,
Columns: idsOrKeys,
FilterFields: fieldFilters,
Limit: limit,
Offset: offset,
})
if err != nil {
return nil, errors.Wrap(err, "getting stream")
} else if stream == nil {

140
executor.go
View file

@ -23,7 +23,7 @@ import (
"sync"
"time"
"github.com/pilosa/pilosa/v2/ext"
"github.com/molecula/ext"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/shardwidth"
@ -527,6 +527,8 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
return e.executeOptionsCall(ctx, index, c, shards, opt)
case "IncludesColumn":
return e.executeIncludesColumnCall(ctx, index, c, shards, opt)
case "All":
return e.executeAllCall(ctx, index, c, shards, opt)
case "Precomputed":
return e.executePrecomputedCall(ctx, index, c, shards, opt)
default:
@ -635,6 +637,112 @@ func (e *executor) executeIncludesColumnCall(ctx context.Context, index string,
return result.(bool), nil
}
// executeAllCall executes an All() call.
func (e *executor) executeAllCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
rslt := NewRow()
var limit uint64
var offset uint64
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, errors.Wrap(err, "getting limit")
} else if hasLimit && lim > 0 {
limit = uint64(lim)
}
if off, hasOffset, err := c.UintArg("offset"); err != nil {
return nil, errors.Wrap(err, "getting offset")
} else if hasOffset && off > 0 {
offset = uint64(off)
}
if limit == 0 {
limit = math.MaxUint64
}
// skip tracks the number of records left to be skipped
// in support of getting to the offset.
var skip uint64 = offset
// got tracks the number of records gotten to that point.
var got uint64
for _, shard := range shards {
row, err := e.executeAllCallMapReduce(ctx, index, c, shard, opt)
if err != nil {
return nil, errors.Wrap(err, "executing map reduce on shard")
}
segCnt := row.Count()
// If this segment doesn't reach the offset, skip it.
if segCnt <= skip {
skip -= segCnt
continue
}
// This segment doesn't have enough to finish fulfilling the limit
// (or it has exactly enough).
if segCnt-skip <= limit-got {
if skip == 0 {
rslt.Merge(row)
} else {
cols := row.Columns()
partialRow := NewRow()
for _, bit := range cols[skip:] {
partialRow.SetBit(bit)
}
rslt.Merge(partialRow)
}
got += segCnt - skip
// In the case where this segment exactly fulfills the limit, break.
if got == limit {
break
}
skip = 0
continue
}
// This segment has more records than the remaining limit requires.
cols := row.Columns()
partialRow := NewRow()
for _, bit := range cols[skip : skip+limit-got] {
partialRow.SetBit(bit)
}
rslt.Merge(partialRow)
break
}
return rslt, nil
}
// executeAllCallMapReduce executes a single shard of the All() call
// using the executor.mapReduce() method.
func (e *executor) executeAllCallMapReduce(ctx context.Context, index string, c *pql.Call, shard uint64, opt *execOptions) (*Row, error) {
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeAllShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
other, _ := prev.(*Row)
if other == nil {
other = NewRow()
}
other.Merge(v.(*Row))
return other
}
result, err := e.mapReduce(ctx, index, []uint64{shard}, c, opt, mapFn, reduceFn)
if err != nil {
return nil, errors.Wrap(err, "map reduce")
}
row, _ := result.(*Row)
return row, nil
}
// executeIncludesColumnCallShard
func (e *executor) executeIncludesColumnCallShard(ctx context.Context, index string, c *pql.Call, shard uint64, column uint64) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeIncludesColumnCallShard")
@ -2381,7 +2489,7 @@ func (e *executor) executePrecomputedCallShard(ctx context.Context, index string
return nil, fmt.Errorf("per-shard: missing precomputed values for shard %d", shard)
}
// executeNotShard executes a not() call for a local shard.
// executeNotShard executes a Not() call for a local shard.
func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeNotShard")
defer span.Finish()
@ -2416,6 +2524,34 @@ func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Cal
return existenceRow.Difference(row), nil
}
// executeAllShard executes an All() call for a local shard.
func (e *executor) executeAllShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeAllShard")
defer span.Finish()
if len(c.Children) > 0 {
return nil, errors.New("All() does not accept an input row")
}
// Make sure the index supports existence tracking.
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
} else if idx.existenceField() == nil {
return nil, errors.Errorf("index does not support existence tracking: %s", index)
}
var existenceRow *Row
existenceFrag := e.Holder.fragment(index, existenceFieldName, viewStandard, shard)
if existenceFrag == nil {
existenceRow = NewRow()
} else {
existenceRow = existenceFrag.row(0)
}
return existenceRow, nil
}
// executeShiftShard executes a shift() call for a local shard.
func (e *executor) executeShiftShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
n, _, err := c.IntArg("n")

143
executor_test.go
View file

@ -2898,6 +2898,149 @@ func TestExecutor_Execute_Not(t *testing.T) {
})
}
// Ensure an all query can be executed.
func TestExecutor_Execute_All(t *testing.T) {
t.Run("ColumnID", func(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
hldr := test.Holder{Holder: c[0].Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{TrackExistence: true})
fld, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
// Create an import request that sets a full shard,
// plus a couple bits set on either side of it, and
// a final bit set in a fourth shard.
//
// shard0 shard1 shard2 shard3
// |----------|----------|----------|----------|
// | **|**********|** | *
//
bitCount := ShardWidth + 5
req := &pilosa.ImportRequest{
Index: index.Name(),
Field: fld.Name(),
Shard: 0,
RowIDs: make([]uint64, bitCount),
ColumnIDs: make([]uint64, bitCount),
}
for i := 0; i < bitCount-1; i++ {
req.RowIDs[i] = 10
req.ColumnIDs[i] = uint64(i + ShardWidth - 2)
}
req.RowIDs[bitCount-1] = 10
req.ColumnIDs[bitCount-1] = uint64((3 * ShardWidth) + 2)
if err := c[0].API.Import(context.Background(), req); err != nil {
t.Fatal(err)
}
tests := []struct {
qry string
expCols []uint64
expCnt uint64
}{
{qry: "All()", expCols: req.ColumnIDs, expCnt: uint64(bitCount)},
{qry: "All(limit=1)", expCols: req.ColumnIDs[:1], expCnt: 1},
{qry: "All(limit=4)", expCols: req.ColumnIDs[:4], expCnt: 4},
{qry: "All(limit=4, offset=4)", expCols: req.ColumnIDs[4:8], expCnt: 4},
{qry: fmt.Sprintf("All(limit=4, offset=%d)", bitCount-5), expCols: req.ColumnIDs[bitCount-5 : bitCount-1], expCnt: 4},
{qry: fmt.Sprintf("All(limit=1, offset=%d)", bitCount-2), expCols: req.ColumnIDs[bitCount-2 : bitCount-1], expCnt: 1},
{qry: fmt.Sprintf("All(limit=1, offset=%d)", bitCount-2), expCols: req.ColumnIDs[bitCount-2 : bitCount-1], expCnt: 1},
{qry: fmt.Sprintf("All(limit=4, offset=%d)", bitCount-2), expCols: req.ColumnIDs[bitCount-2:], expCnt: 2},
{qry: fmt.Sprintf("All(limit=4, offset=%d)", bitCount+1), expCols: []uint64{}, expCnt: 0},
{qry: fmt.Sprintf("All(limit=2, offset=%d)", bitCount-3), expCols: req.ColumnIDs[bitCount-3 : bitCount-1], expCnt: 2},
{qry: fmt.Sprintf("All(limit=2, offset=%d)", bitCount-5), expCols: req.ColumnIDs[bitCount-5 : bitCount-3], expCnt: 2},
{qry: "All(limit=2, offset=2)", expCols: req.ColumnIDs[2:4], expCnt: 2},
{qry: "All(limit=1, offset=1)", expCols: req.ColumnIDs[1:2], expCnt: 1},
{qry: fmt.Sprintf("All(limit=%d, offset=2)", ShardWidth), expCols: req.ColumnIDs[2 : bitCount-3], expCnt: ShardWidth},
}
for i, test := range tests {
if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.qry}); err != nil {
t.Fatal(err)
} else if cnt := res.Results[0].(*pilosa.Row).Count(); cnt != test.expCnt {
t.Fatalf("test %d, unexpected count, got: %d, but expected: %d", i, cnt, test.expCnt)
} else if cols := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(cols, test.expCols) {
// If the error results are too large, just show the count.
if len(cols) > 1000 || len(test.expCols) > 1000 {
t.Fatalf("test %d, unexpected columns, got: len(%d), but expected: len(%d)", i, len(cols), len(test.expCols))
} else {
t.Fatalf("test %d, unexpected columns, got: %v, but expected: %v", i, cols, test.expCols)
}
}
}
})
t.Run("ColumnKey", func(t *testing.T) {
c := test.MustRunCluster(t, 1, []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
),
})
defer c.Close()
hldr := test.Holder{Holder: c[0].Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{TrackExistence: true, Keys: true})
fld, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
// Create an import request that sets key columns
//
// shard0
// |----------|
// |**** |
//
bitCount := 4
req := &pilosa.ImportRequest{
Index: index.Name(),
Field: fld.Name(),
Shard: 0,
RowIDs: make([]uint64, bitCount),
ColumnKeys: make([]string, bitCount),
}
for i := 0; i < bitCount; i++ {
req.RowIDs[i] = 10
req.ColumnKeys[i] = fmt.Sprintf("c%d", i)
}
if err := c[0].API.Import(context.Background(), req); err != nil {
t.Fatal(err)
}
tests := []struct {
qry string
expCols []string
expCnt uint64
}{
{qry: "All()", expCols: req.ColumnKeys, expCnt: uint64(bitCount)},
{qry: "All(limit=1)", expCols: req.ColumnKeys[:1], expCnt: 1},
{qry: "All(limit=4)", expCols: req.ColumnKeys, expCnt: 4},
{qry: "All(limit=5)", expCols: req.ColumnKeys, expCnt: 4},
{qry: "All(limit=1, offset=1)", expCols: req.ColumnKeys[1:2], expCnt: 1},
{qry: "All(limit=4, offset=1)", expCols: req.ColumnKeys[1:], expCnt: 3},
{qry: "All(limit=4, offset=5)", expCols: nil, expCnt: 0},
}
for i, test := range tests {
if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.qry}); err != nil {
t.Fatal(err)
} else if cnt := len(res.Results[0].(*pilosa.Row).Keys); uint64(cnt) != test.expCnt {
t.Fatalf("test %d, unexpected count, got: %d, but expected: %d", i, cnt, test.expCnt)
} else if cols := res.Results[0].(*pilosa.Row).Keys; !reflect.DeepEqual(cols, test.expCols) {
// If the error results are too large, just show the count.
if len(cols) > 1000 || len(test.expCols) > 1000 {
t.Fatalf("test %d, unexpected columns, got: len(%d), but expected: len(%d)", i, len(cols), len(test.expCols))
} else {
t.Fatalf("test %d, unexpected columns, got: %T, but expected: %T", i, cols, test.expCols)
}
}
}
})
}
// Ensure a row can be cleared.
func TestExecutor_Execute_ClearRow(t *testing.T) {
// Set and Mutex tests use the same data and queries

238
ext/ext.go
View file

@ -1,238 +0,0 @@
// Copyright 2019 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 ext provides an EXPERIMENTAL AND TEMPORARY interface to use for
// plugin extensions to Pilosa. DO NOT DEVELOP NEW PLUGINS WITH THIS. The
// replacement design is already in process, but it needs more refinement
// to address issues. This one has those issues, and more.
//
// In the current design, plugins will be loaded at runtime using the
// go `plugin` package, so they should be built as a main package using
// the plugin build mode.
//
// Plugins should not import other packages from Pilosa.
//
// To advertise their functionality, plugins define one or more of a
// handful of symbols which will be checked for at plugin load and used
// to register their functionality.
//
// The plugin interface will check for the following function(s). If the
// functions exist, they must have the given signatures. If they return
// a non-nil error, no ops are registered, and the error message will
// be reported in the Pilosa server's logs.
//
// BitmapOps() ([]BitmapOp, error)
//
// These functions may be absent, and may return nil slices; in either
// case, no ops are registered.
package ext
// The Bitmap type represents a Pilosa bitmap, and is used for bitmap
// operations.
type Bitmap interface {
// AddN and RemoveN can be used to add or remove values from a bitmap.
AddN(a ...uint64) (int, error)
RemoveN(a ...uint64) (int, error)
// Lookups
Max() uint64
Min() (uint64, bool)
Count() uint64
Any() bool
Contains(uint64) bool
Slice() []uint64
SliceRange(uint64, uint64) []uint64
// ContainerBits stores the next 1<<16 bits, starting at the provided
// bit index. It may use a provided []uint64 to store them, or may
// provide its own. Don't write to those bits. Offset must be a multiple
// of 1<<16.
ContainerBits(uint64, []uint64) []uint64
// These operators provide existing implemented binary ops.
Intersect(Bitmap) Bitmap
Union(Bitmap) Bitmap
IntersectionCount(Bitmap) uint64
Difference(Bitmap) Bitmap
Xor(Bitmap) Bitmap
Shift(int) (Bitmap, error)
Flip(uint64, uint64) Bitmap
// New() is an atrocity: it creates a new bitmap, unrelated to the
// existing bitmap. This lets you create a new bitmap without having
// imported any of the packages that have bitmap creation tools, because
// the bitmap wrapper type has to give you one.
New() Bitmap
}
// SignedBitmap represents a bitmap that can contain both positive and negative
// values.
type SignedBitmap struct {
Pos, Neg Bitmap
}
// A BitmapOp represents a new bitmap operation that should be exposed
// in PQL.
type BitmapOpInput byte
type BitmapOpOutput byte
type BitmapOpArity byte
type BitmapOpPrecall byte
type BitmapOpType struct {
Input BitmapOpInput
Arity BitmapOpArity
Output BitmapOpOutput
Precall BitmapOpPrecall
}
const (
OpArityUnary = BitmapOpArity(iota)
OpArityBinary
OpArityNary
)
const (
// Unary: Exactly one bitmap.
OpInputBitmap = BitmapOpInput(iota)
// The really weird special case used for BSI, where we end up
// needing to do BSI computations. Arguments will be a
// single BitmapBSI, and a []Bitmap for other operands if any.
OpInputNaryBSI
)
const (
OpOutputCount = BitmapOpOutput(iota)
OpOutputBitmap
OpOutputSignedBitmap
)
const (
OpPrecallNone = BitmapOpPrecall(iota)
OpPrecallGlobal
OpPrecallLocal // unimplemented
)
// Regardless of arity, non-BSI functions should always take []Bitmap.
type BitmapOpFunc interface {
BitmapOpType() BitmapOpType
}
// BitmapOpBitmap should actually always be func([]Bitmap) Bitmap, but
// might be different kinds.
type BitmapOpBitmap interface {
BitmapOpArity() BitmapOpArity
BitmapOpFunc() GenericBitmapOpBitmap
}
// the common underlying type of the other BitmapOpBitmap functions
type GenericBitmapOpBitmap func([]Bitmap, map[string]interface{}) Bitmap
// BitmapBSI represents the way a single BSI field is passed into a function
// which takes a BSI field.
type BitmapBSI struct {
FieldData Bitmap
ShardWidth uint64
Offset int64
Depth uint
}
type BitmapOpBSIBitmap func(BitmapBSI, []Bitmap, map[string]interface{}) SignedBitmap
func (b BitmapOpBSIBitmap) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputNaryBSI, Arity: OpArityNary, Output: OpOutputSignedBitmap}
}
type BitmapOpBSIBitmapPrecall func(BitmapBSI, []Bitmap, map[string]interface{}) SignedBitmap
func (b BitmapOpBSIBitmapPrecall) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputNaryBSI, Arity: OpArityNary, Precall: OpPrecallGlobal, Output: OpOutputSignedBitmap}
}
type BitmapOpUnaryCount func([]Bitmap, map[string]interface{}) int64
func (b BitmapOpUnaryCount) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputBitmap, Arity: OpArityUnary, Output: OpOutputCount}
}
type BitmapOpUnaryBitmap func([]Bitmap, map[string]interface{}) Bitmap
func (b BitmapOpUnaryBitmap) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputBitmap, Arity: OpArityUnary, Output: OpOutputBitmap}
}
func (b BitmapOpUnaryBitmap) BitmapOpArity() BitmapOpArity {
return OpArityUnary
}
func (b BitmapOpUnaryBitmap) BitmapOpFunc() GenericBitmapOpBitmap {
return GenericBitmapOpBitmap(b)
}
type BitmapOpBinaryBitmap func([]Bitmap, map[string]interface{}) Bitmap
func (b BitmapOpBinaryBitmap) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputBitmap, Arity: OpArityBinary, Output: OpOutputBitmap}
}
func (b BitmapOpBinaryBitmap) BitmapOpArity() BitmapOpArity {
return OpArityBinary
}
func (b BitmapOpBinaryBitmap) BitmapOpFunc() GenericBitmapOpBitmap {
return GenericBitmapOpBitmap(b)
}
type BitmapOpNaryBitmap func([]Bitmap, map[string]interface{}) Bitmap
func (b BitmapOpNaryBitmap) BitmapOpType() BitmapOpType {
return BitmapOpType{Input: OpInputBitmap, Arity: OpArityNary, Output: OpOutputBitmap}
}
func (b BitmapOpNaryBitmap) BitmapOpArity() BitmapOpArity {
return OpArityNary
}
func (b BitmapOpNaryBitmap) BitmapOpFunc() GenericBitmapOpBitmap {
return GenericBitmapOpBitmap(b)
}
// BitmapOp represents an operation to be supported in PQL. Operations
// on bitmaps should always take []Bitmap. Operations on InputNaryBSI should
// take a []Bitmap, plus a Bitmap/shard-width/offset/depth.
//
// Reserved is a list of words to treat as reserved words in a prototype.
// This is not currently used but might be later, and I want to have the
// concept handy now.
type BitmapOp struct {
Name string
Func BitmapOpFunc
Reserved []string
}
// ExtensionInfo tells us about the extension. The ExtensionAPI string
// should be "v0". The version is a human-readable version, use something
// that seems meaningful. Name and Description are reasonably self-explanatory,
// I hope.
//
// Extensions should define a function:
// func ExtensionInfo(extensionAPI string) (*ExtensionInfo, error)
// which reports their extension info if they think they can coexist with that
// API string.
type ExtensionInfo struct {
Name string // Extension name.
Description string // Short description.
Version string // Human-readable version info for extension.
ExtensionAPI string // Extension API version. Should be v0 for now.
License string // License info.
BitmapOps []BitmapOp // List of provided ops.
}

1
ext/samples/.gitignore vendored
View file

@ -1 +0,0 @@
*/*.so

125
ext/samples/some/some.go
View file

@ -1,125 +0,0 @@
// Copyright 2019 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 main
import (
"fmt"
"math/bits"
"github.com/molecula/apophenia"
"github.com/pilosa/pilosa/v2/ext"
)
// This could be dynamically generated, but for now it's not.
// nolint:unused,deadcode
var extInfoTemplate = &ext.ExtensionInfo{
Name: "some",
Description: "some of the bits/all of the bits/none of the bits",
Version: "0.01",
ExtensionAPI: "v0",
License: "unreleased",
BitmapOps: []ext.BitmapOp{
{Name: "Some", Func: ext.BitmapOpUnaryBitmap(Some), Reserved: []string{"p", "seed"}},
},
}
// ExtensionInfo is the entry point used by the plugin code.
func ExtensionInfo(api string) (*ext.ExtensionInfo, error) { // nolint:unused,deadcode
return extInfoTemplate, nil
}
const batchSize = 1024
// Some returns some of the bits from its first input bitmap. Takes seed (int)
// and p (float) values. Seed defaults to 0.
func Some(inputs []ext.Bitmap, args map[string]interface{}) ext.Bitmap {
if len(inputs) == 0 || inputs[0] == nil {
return nil
}
input := inputs[0]
min, ok := input.Min()
// no bits found?
if !ok {
return nil
}
// start at multiple of 128 not greater than min.
min &^= 127
max := input.Max()
p, ok := args["p"].(float64)
if !ok {
return nil
}
// no bits or impossible probability range
if p <= 0 || p > 1 {
return nil
}
// every bit
if p == 1 {
return inputs[0]
}
// On failure, we default to 0.
seed, _ := args["seed"].(int64)
densityScale := uint64(256)
density := uint64(p * float64(densityScale))
for density == 0 {
densityScale <<= 1
density = uint64(p * float64(densityScale))
// too small
if densityScale > (1 << 32) {
return nil
}
}
w, err := apophenia.NewWeighted(apophenia.NewSequence(seed))
if err != nil {
return nil
}
someBits := input.New()
toAdd := make([]uint64, batchSize)
toAddN := 0
offset := apophenia.OffsetFor(apophenia.SequenceWeighted, 0, 0, 0)
for i := min; i < max; i += 128 {
offset.Lo = i
randomBits := w.Bits(offset, density, densityScale)
bit := uint64(0)
for randomBits.Lo != 0 {
next := uint64(bits.TrailingZeros64(randomBits.Lo) + 1)
randomBits.Lo >>= next
toAdd[toAddN] = next + bit + i
toAddN++
bit += next
}
bit = 64
for randomBits.Hi != 0 {
next := uint64(bits.TrailingZeros64(randomBits.Hi) + 1)
randomBits.Hi >>= next
toAdd[toAddN] = next + bit + i
toAddN++
bit += next
}
if toAddN > (batchSize - 128) {
// ignore error
_, _ = someBits.AddN(toAdd[:toAddN]...)
toAddN = 0
}
}
if toAddN > 0 {
_, _ = someBits.AddN(toAdd[:toAddN]...)
}
return input.Intersect(someBits)
}
func main() {
fmt.Printf("this is a plugin module only.\n")
}

2
extension.go
View file

@ -17,7 +17,7 @@ package pilosa
import (
"fmt"
"github.com/pilosa/pilosa/v2/ext"
"github.com/molecula/ext"
"github.com/pilosa/pilosa/v2/roaring"
)

21
extensions/distinct.go Normal file
View file

@ -0,0 +1,21 @@
// Copyright 2019 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.
// +build plugindistinct
package extensions
import (
_ "github.com/molecula/extensions/distinct"
)

18
extensions/dummy.go Normal file
View file

@ -0,0 +1,18 @@
// Copyright 2019 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.
// This package contains only things which are conditional on build
// tags.
package extensions

7
go.mod
View file

@ -18,7 +18,8 @@ require (
github.com/gorilla/mux v1.7.0
github.com/hashicorp/memberlist v0.1.3
github.com/inconshreveable/mousetrap v1.0.0 // indirect
github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b
github.com/molecula/ext v0.0.0-20191202195653-240f38a75171
github.com/molecula/extensions v0.0.0-20191218165536-562244600fd4
github.com/opentracing/opentracing-go v1.1.0
github.com/pelletier/go-toml v1.2.0
github.com/pkg/errors v0.8.1
@ -34,15 +35,17 @@ require (
github.com/uber-go/atomic v1.4.0 // indirect
github.com/uber/jaeger-client-go v2.16.0+incompatible
github.com/uber/jaeger-lib v2.2.0+incompatible // indirect
github.com/youtube/vitess v2.1.1+incompatible // indirect
go.uber.org/atomic v1.4.0 // indirect
golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 // indirect
golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect
golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6
golang.org/x/sync v0.0.0-20190423024810-112230192c58
golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872 // indirect
golang.org/x/text v0.3.2 // indirect
google.golang.org/grpc v1.24.0
modernc.org/mathutil v1.0.0
modernc.org/strutil v1.0.0
vitess.io/vitess v2.1.1+incompatible // indirect
)
go 1.13

8
go.sum
View file

@ -89,6 +89,10 @@ github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQz
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b h1:cZADDaNYM7xn/nklO3g198JerGQjadFuA0ofxBJgK0Y=
github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b/go.mod h1:uXd1BiH7xLmgkhVmspdJLENv6uGWrTL/MQX2TN7Yz9s=
github.com/molecula/ext v0.0.0-20191202195653-240f38a75171 h1:4VK7u/RM+54Yaz8aRB9vIaDSnbKi3M0NQYg5tsZvOT4=
github.com/molecula/ext v0.0.0-20191202195653-240f38a75171/go.mod h1:r6EIj0GH8dx5xxFLW6Voi1/mX3wXOUkJu6AoEE/xvGQ=
github.com/molecula/extensions v0.0.0-20191218165536-562244600fd4 h1:mDB/dicofRVFuRYcCVPk+JBiVKXlfbzMahuqHvrYqu4=
github.com/molecula/extensions v0.0.0-20191218165536-562244600fd4/go.mod h1:QQgN5OFjuBAi4Q2UYVMzfvi4k9yvg/qqC+MNFB4I9JI=
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
@ -153,6 +157,8 @@ github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/
github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
github.com/youtube/vitess v2.1.1+incompatible h1:SE+P7DNX/jw5RHFs5CHRhZQjq402EJFCD33JhzQMdDw=
github.com/youtube/vitess v2.1.1+incompatible/go.mod h1:hpMim5/30F1r+0P8GGtB29d0gWHr0IZ5unS+CG0zMx8=
go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@ -210,3 +216,5 @@ modernc.org/mathutil v1.0.0 h1:93vKjrJopTPrtTNpZ8XIovER7iCIH1QU7wNbOQXC60I=
modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
modernc.org/strutil v1.0.0 h1:XVFtQwFVwc02Wk+0L/Z/zDDXO81r5Lhe6iMKmGX3KhE=
modernc.org/strutil v1.0.0/go.mod h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs=
vitess.io/vitess v2.1.1+incompatible h1:nuuGHiWYWpudD3gOCLeGzol2EJ25e/u5Wer2wV1O130=
vitess.io/vitess v2.1.1+incompatible/go.mod h1:h4qvkyNYTOC0xI+vcidSWoka0gQAZc9ZPHbkHo48gP0=

2
internal/clustertests/docker-compose.yml
View file

@ -51,6 +51,6 @@ services:
volumes:
- /var/run/docker.sock:/var/run/docker.sock
command:
- "cd /go/src/github.com/pilosa/pilosa/ && go test -v -count=1 github.com/pilosa/pilosa/v2/internal/clustertests"
- "cd /go/src/github.com/pilosa/pilosa/ && go test -mod=vendor -v -count=1 github.com/pilosa/pilosa/v2/internal/clustertests"
networks:
pilosanet:

17
pql/ast.go
View file

@ -23,7 +23,7 @@ import (
"strings"
"time"
"github.com/pilosa/pilosa/v2/ext"
"github.com/molecula/ext"
)
// Query represents a PQL query.
@ -347,10 +347,17 @@ var callInfoByFunc = map[string]callInfo{
"Difference": {allowUnknown: false},
"Intersect": {allowUnknown: false},
"Not": {allowUnknown: false},
"ClearRow": {allowUnknown: true},
"Store": {allowUnknown: true},
"MinRow": allowField,
"MaxRow": allowField,
"All": {
allowUnknown: false,
prototypes: map[string]interface{}{
"limit": int64(0),
"offset": int64(0),
},
},
"ClearRow": {allowUnknown: true},
"Store": {allowUnknown: true},
"MinRow": allowField,
"MaxRow": allowField,
"Rows": {
allowUnknown: false,
prototypes: map[string]interface{}{

343
proto/pilosa.pb.go
View file

@ -4,15 +4,16 @@
package pilosa
import (
context "context"
fmt "fmt"
proto "github.com/golang/protobuf/proto"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
math "math"
)
import (
context "golang.org/x/net/context"
grpc "google.golang.org/grpc"
)
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
@ -22,7 +23,7 @@ var _ = math.Inf
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
type QueryPQLRequest struct {
Index string `protobuf:"bytes,1,opt,name=index,proto3" json:"index,omitempty"`
@ -321,9 +322,9 @@ func (m *ColumnResponse) GetFloat64Val() float64 {
return 0
}
// XXX_OneofWrappers is for the internal use of the proto package.
func (*ColumnResponse) XXX_OneofWrappers() []interface{} {
return []interface{}{
// XXX_OneofFuncs is for the internal use of the proto package.
func (*ColumnResponse) XXX_OneofFuncs() (func(msg proto.Message, b *proto.Buffer) error, func(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error), func(msg proto.Message) (n int), []interface{}) {
return _ColumnResponse_OneofMarshaler, _ColumnResponse_OneofUnmarshaler, _ColumnResponse_OneofSizer, []interface{}{
(*ColumnResponse_StringVal)(nil),
(*ColumnResponse_Uint64Val)(nil),
(*ColumnResponse_Int64Val)(nil),
@ -335,10 +336,162 @@ func (*ColumnResponse) XXX_OneofWrappers() []interface{} {
}
}
func _ColumnResponse_OneofMarshaler(msg proto.Message, b *proto.Buffer) error {
m := msg.(*ColumnResponse)
// columnVal
switch x := m.ColumnVal.(type) {
case *ColumnResponse_StringVal:
b.EncodeVarint(1<<3 | proto.WireBytes)
b.EncodeStringBytes(x.StringVal)
case *ColumnResponse_Uint64Val:
b.EncodeVarint(2<<3 | proto.WireVarint)
b.EncodeVarint(uint64(x.Uint64Val))
case *ColumnResponse_Int64Val:
b.EncodeVarint(3<<3 | proto.WireVarint)
b.EncodeVarint(uint64(x.Int64Val))
case *ColumnResponse_BoolVal:
t := uint64(0)
if x.BoolVal {
t = 1
}
b.EncodeVarint(4<<3 | proto.WireVarint)
b.EncodeVarint(t)
case *ColumnResponse_BlobVal:
b.EncodeVarint(5<<3 | proto.WireBytes)
b.EncodeRawBytes(x.BlobVal)
case *ColumnResponse_Uint64ArrayVal:
b.EncodeVarint(6<<3 | proto.WireBytes)
if err := b.EncodeMessage(x.Uint64ArrayVal); err != nil {
return err
}
case *ColumnResponse_StringArrayVal:
b.EncodeVarint(7<<3 | proto.WireBytes)
if err := b.EncodeMessage(x.StringArrayVal); err != nil {
return err
}
case *ColumnResponse_Float64Val:
b.EncodeVarint(8<<3 | proto.WireFixed64)
b.EncodeFixed64(math.Float64bits(x.Float64Val))
case nil:
default:
return fmt.Errorf("ColumnResponse.ColumnVal has unexpected type %T", x)
}
return nil
}
func _ColumnResponse_OneofUnmarshaler(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error) {
m := msg.(*ColumnResponse)
switch tag {
case 1: // columnVal.stringVal
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeStringBytes()
m.ColumnVal = &ColumnResponse_StringVal{x}
return true, err
case 2: // columnVal.uint64Val
if wire != proto.WireVarint {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeVarint()
m.ColumnVal = &ColumnResponse_Uint64Val{x}
return true, err
case 3: // columnVal.int64Val
if wire != proto.WireVarint {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeVarint()
m.ColumnVal = &ColumnResponse_Int64Val{int64(x)}
return true, err
case 4: // columnVal.boolVal
if wire != proto.WireVarint {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeVarint()
m.ColumnVal = &ColumnResponse_BoolVal{x != 0}
return true, err
case 5: // columnVal.blobVal
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeRawBytes(true)
m.ColumnVal = &ColumnResponse_BlobVal{x}
return true, err
case 6: // columnVal.uint64ArrayVal
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
msg := new(Uint64Array)
err := b.DecodeMessage(msg)
m.ColumnVal = &ColumnResponse_Uint64ArrayVal{msg}
return true, err
case 7: // columnVal.stringArrayVal
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
msg := new(StringArray)
err := b.DecodeMessage(msg)
m.ColumnVal = &ColumnResponse_StringArrayVal{msg}
return true, err
case 8: // columnVal.float64Val
if wire != proto.WireFixed64 {
return true, proto.ErrInternalBadWireType
}
x, err := b.DecodeFixed64()
m.ColumnVal = &ColumnResponse_Float64Val{math.Float64frombits(x)}
return true, err
default:
return false, nil
}
}
func _ColumnResponse_OneofSizer(msg proto.Message) (n int) {
m := msg.(*ColumnResponse)
// columnVal
switch x := m.ColumnVal.(type) {
case *ColumnResponse_StringVal:
n += 1 // tag and wire
n += proto.SizeVarint(uint64(len(x.StringVal)))
n += len(x.StringVal)
case *ColumnResponse_Uint64Val:
n += 1 // tag and wire
n += proto.SizeVarint(uint64(x.Uint64Val))
case *ColumnResponse_Int64Val:
n += 1 // tag and wire
n += proto.SizeVarint(uint64(x.Int64Val))
case *ColumnResponse_BoolVal:
n += 1 // tag and wire
n += 1
case *ColumnResponse_BlobVal:
n += 1 // tag and wire
n += proto.SizeVarint(uint64(len(x.BlobVal)))
n += len(x.BlobVal)
case *ColumnResponse_Uint64ArrayVal:
s := proto.Size(x.Uint64ArrayVal)
n += 1 // tag and wire
n += proto.SizeVarint(uint64(s))
n += s
case *ColumnResponse_StringArrayVal:
s := proto.Size(x.StringArrayVal)
n += 1 // tag and wire
n += proto.SizeVarint(uint64(s))
n += s
case *ColumnResponse_Float64Val:
n += 1 // tag and wire
n += 8
case nil:
default:
panic(fmt.Sprintf("proto: unexpected type %T in oneof", x))
}
return n
}
type InspectRequest struct {
Index string `protobuf:"bytes,1,opt,name=index,proto3" json:"index,omitempty"`
Columns *IdsOrKeys `protobuf:"bytes,2,opt,name=columns,proto3" json:"columns,omitempty"`
FilterFields []string `protobuf:"bytes,3,rep,name=filterFields,proto3" json:"filterFields,omitempty"`
Limit uint64 `protobuf:"varint,4,opt,name=limit,proto3" json:"limit,omitempty"`
Offset uint64 `protobuf:"varint,5,opt,name=offset,proto3" json:"offset,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -390,6 +543,20 @@ func (m *InspectRequest) GetFilterFields() []string {
return nil
}
func (m *InspectRequest) GetLimit() uint64 {
if m != nil {
return m.Limit
}
return 0
}
func (m *InspectRequest) GetOffset() uint64 {
if m != nil {
return m.Offset
}
return 0
}
type Uint64Array struct {
Vals []uint64 `protobuf:"varint,1,rep,packed,name=vals,proto3" json:"vals,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
@ -540,14 +707,80 @@ func (m *IdsOrKeys) GetKeys() *StringArray {
return nil
}
// XXX_OneofWrappers is for the internal use of the proto package.
func (*IdsOrKeys) XXX_OneofWrappers() []interface{} {
return []interface{}{
// XXX_OneofFuncs is for the internal use of the proto package.
func (*IdsOrKeys) XXX_OneofFuncs() (func(msg proto.Message, b *proto.Buffer) error, func(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error), func(msg proto.Message) (n int), []interface{}) {
return _IdsOrKeys_OneofMarshaler, _IdsOrKeys_OneofUnmarshaler, _IdsOrKeys_OneofSizer, []interface{}{
(*IdsOrKeys_Ids)(nil),
(*IdsOrKeys_Keys)(nil),
}
}
func _IdsOrKeys_OneofMarshaler(msg proto.Message, b *proto.Buffer) error {
m := msg.(*IdsOrKeys)
// type
switch x := m.Type.(type) {
case *IdsOrKeys_Ids:
b.EncodeVarint(1<<3 | proto.WireBytes)
if err := b.EncodeMessage(x.Ids); err != nil {
return err
}
case *IdsOrKeys_Keys:
b.EncodeVarint(2<<3 | proto.WireBytes)
if err := b.EncodeMessage(x.Keys); err != nil {
return err
}
case nil:
default:
return fmt.Errorf("IdsOrKeys.Type has unexpected type %T", x)
}
return nil
}
func _IdsOrKeys_OneofUnmarshaler(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error) {
m := msg.(*IdsOrKeys)
switch tag {
case 1: // type.ids
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
msg := new(Uint64Array)
err := b.DecodeMessage(msg)
m.Type = &IdsOrKeys_Ids{msg}
return true, err
case 2: // type.keys
if wire != proto.WireBytes {
return true, proto.ErrInternalBadWireType
}
msg := new(StringArray)
err := b.DecodeMessage(msg)
m.Type = &IdsOrKeys_Keys{msg}
return true, err
default:
return false, nil
}
}
func _IdsOrKeys_OneofSizer(msg proto.Message) (n int) {
m := msg.(*IdsOrKeys)
// type
switch x := m.Type.(type) {
case *IdsOrKeys_Ids:
s := proto.Size(x.Ids)
n += 1 // tag and wire
n += proto.SizeVarint(uint64(s))
n += s
case *IdsOrKeys_Keys:
s := proto.Size(x.Keys)
n += 1 // tag and wire
n += proto.SizeVarint(uint64(s))
n += s
case nil:
default:
panic(fmt.Sprintf("proto: unexpected type %T in oneof", x))
}
return n
}
func init() {
proto.RegisterType((*QueryPQLRequest)(nil), "pilosa.QueryPQLRequest")
proto.RegisterType((*RowResponse)(nil), "pilosa.RowResponse")
@ -559,44 +792,6 @@ func init() {
proto.RegisterType((*IdsOrKeys)(nil), "pilosa.IdsOrKeys")
}
func init() { proto.RegisterFile("pilosa.proto", fileDescriptor_ef0691a44d1e275c) }
var fileDescriptor_ef0691a44d1e275c = []byte{
// 497 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x7c, 0x53, 0x5d, 0x6b, 0xd4, 0x40,
0x14, 0xcd, 0x34, 0xe9, 0xee, 0xe6, 0x66, 0x59, 0xf5, 0x2a, 0x1a, 0x16, 0x91, 0x98, 0x17, 0x23,
0x4a, 0x29, 0xab, 0x08, 0x4a, 0x7d, 0xb0, 0x82, 0x64, 0x51, 0xb0, 0x1d, 0xb1, 0xef, 0xb3, 0xcd,
0x6c, 0x0d, 0xce, 0x66, 0xd2, 0x4c, 0xd6, 0x9a, 0x57, 0xff, 0xa2, 0x7f, 0x48, 0x66, 0xf2, 0xb1,
0x49, 0x61, 0x7d, 0x9b, 0x39, 0xe7, 0xdc, 0xef, 0x7b, 0x61, 0x9a, 0xa7, 0x42, 0x2a, 0x76, 0x94,
0x17, 0xb2, 0x94, 0x38, 0xaa, 0x7f, 0xe1, 0x5b, 0xb8, 0x73, 0xbe, 0xe5, 0x45, 0x75, 0x76, 0xfe,
0x85, 0xf2, 0xeb, 0x2d, 0x57, 0x25, 0x3e, 0x80, 0xc3, 0x34, 0x4b, 0xf8, 0x6f, 0x9f, 0x04, 0x24,
0x72, 0x69, 0xfd, 0xc1, 0xbb, 0x60, 0xe7, 0xd7, 0xc2, 0x3f, 0x30, 0x98, 0x7e, 0x86, 0x1b, 0xf0,
0xa8, 0xbc, 0xa1, 0x5c, 0xe5, 0x32, 0x53, 0x1c, 0x5f, 0xc2, 0xf8, 0x07, 0x67, 0x09, 0x2f, 0x94,
0x4f, 0x02, 0x3b, 0xf2, 0x16, 0x78, 0xd4, 0x44, 0xfc, 0x28, 0xc5, 0x76, 0x93, 0x2d, 0xb3, 0xb5,
0xa4, 0xad, 0x04, 0x8f, 0x61, 0x7c, 0x69, 0x60, 0xe5, 0x1f, 0x18, 0xf5, 0xc3, 0xa1, 0xba, 0x75,
0x4b, 0x5b, 0x59, 0x78, 0x02, 0xb0, 0x73, 0x84, 0x08, 0x4e, 0xc6, 0x36, 0xbc, 0xc9, 0xd1, 0xbc,
0x71, 0x0e, 0x93, 0x84, 0x95, 0xac, 0xac, 0x72, 0xde, 0xe4, 0xd9, 0xfd, 0xc3, 0xbf, 0x07, 0x30,
0x1b, 0x7a, 0xc6, 0x27, 0xe0, 0xaa, 0xb2, 0x48, 0xb3, 0xab, 0x0b, 0x26, 0x6a, 0x3f, 0xb1, 0x45,
0x77, 0x90, 0xe6, 0xb7, 0x69, 0x56, 0xbe, 0x79, 0xad, 0x79, 0xed, 0xcf, 0xd1, 0x7c, 0x07, 0xe1,
0x63, 0x98, 0x74, 0xb4, 0x1d, 0x90, 0xc8, 0x8e, 0x2d, 0xda, 0x21, 0x38, 0x87, 0xf1, 0x4a, 0x4a,
0xa1, 0x49, 0x27, 0x20, 0xd1, 0x24, 0xb6, 0x68, 0x0b, 0x18, 0x4e, 0xc8, 0x95, 0xe6, 0x0e, 0x03,
0x12, 0x4d, 0x0d, 0x57, 0x03, 0xf8, 0x1e, 0x66, 0x75, 0x88, 0x0f, 0x45, 0xc1, 0x2a, 0x2d, 0x19,
0x05, 0x24, 0xf2, 0x16, 0xf7, 0xdb, 0xfe, 0x7c, 0xdf, 0xb1, 0xb1, 0x45, 0x6f, 0x89, 0xb5, 0x79,
0x5d, 0x41, 0x67, 0x3e, 0x1e, 0x9a, 0x7f, 0xdb, 0xb1, 0xda, 0x7c, 0x28, 0xc6, 0x00, 0x60, 0x2d,
0x24, 0x6b, 0xaa, 0x9a, 0x04, 0x24, 0x22, 0xb1, 0x45, 0x7b, 0xd8, 0xa9, 0x07, 0x6e, 0x3d, 0x91,
0x0b, 0x26, 0xc2, 0x1b, 0x98, 0x2d, 0x33, 0x95, 0xf3, 0xcb, 0xf2, 0xff, 0xcb, 0xf3, 0xa2, 0x3f,
0x6d, 0x9d, 0xce, 0xbd, 0x36, 0x9d, 0x65, 0xa2, 0xbe, 0x16, 0x9f, 0x79, 0xa5, 0xba, 0x41, 0x63,
0x08, 0xd3, 0x75, 0x2a, 0x4a, 0x5e, 0x7c, 0x4a, 0xb9, 0x48, 0x94, 0x6f, 0x07, 0x76, 0xe4, 0xd2,
0x01, 0x16, 0x3e, 0x05, 0xaf, 0xd7, 0x07, 0xbd, 0x0d, 0xbf, 0x98, 0xa8, 0x17, 0xcf, 0xa1, 0xe6,
0xad, 0x25, 0xbd, 0x5a, 0x07, 0x12, 0xb7, 0x91, 0x5c, 0x81, 0xdb, 0xc5, 0xc7, 0x67, 0x60, 0xa7,
0x89, 0x32, 0x79, 0xef, 0xed, 0xb6, 0x56, 0xe0, 0x73, 0x70, 0x7e, 0xf2, 0xaa, 0xad, 0x64, 0x4f,
0x63, 0x8d, 0xe4, 0x74, 0x04, 0x8e, 0xde, 0xbe, 0xc5, 0x1f, 0x02, 0xa3, 0x33, 0x23, 0xc3, 0x13,
0x98, 0xb4, 0x07, 0x87, 0x8f, 0x5a, 0xdb, 0x5b, 0x27, 0x38, 0xef, 0x9c, 0xf6, 0x0e, 0x2c, 0xb4,
0x8e, 0x09, 0xbe, 0x83, 0x71, 0xd3, 0x70, 0xec, 0x0e, 0x66, 0x38, 0x81, 0xbd, 0xb6, 0xab, 0x91,
0xb9, 0xfc, 0x57, 0xff, 0x02, 0x00, 0x00, 0xff, 0xff, 0xca, 0x68, 0x70, 0x08, 0x09, 0x04, 0x00,
0x00,
}
// Reference imports to suppress errors if they are not otherwise used.
var _ context.Context
var _ grpc.ClientConn
@ -691,17 +886,6 @@ type PilosaServer interface {
Inspect(*InspectRequest, Pilosa_InspectServer) error
}
// UnimplementedPilosaServer can be embedded to have forward compatible implementations.
type UnimplementedPilosaServer struct {
}
func (*UnimplementedPilosaServer) QueryPQL(req *QueryPQLRequest, srv Pilosa_QueryPQLServer) error {
return status.Errorf(codes.Unimplemented, "method QueryPQL not implemented")
}
func (*UnimplementedPilosaServer) Inspect(req *InspectRequest, srv Pilosa_InspectServer) error {
return status.Errorf(codes.Unimplemented, "method Inspect not implemented")
}
func RegisterPilosaServer(s *grpc.Server, srv PilosaServer) {
s.RegisterService(&_Pilosa_serviceDesc, srv)
}
@ -766,3 +950,42 @@ var _Pilosa_serviceDesc = grpc.ServiceDesc{
},
Metadata: "pilosa.proto",
}
func init() { proto.RegisterFile("pilosa.proto", fileDescriptor_ef0691a44d1e275c) }
var fileDescriptor_ef0691a44d1e275c = []byte{
// 524 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x7c, 0x54, 0xdd, 0x8a, 0xd3, 0x40,
0x14, 0xce, 0x34, 0xd9, 0xb4, 0x39, 0x2d, 0x55, 0x8f, 0xb2, 0x96, 0x22, 0x12, 0x73, 0x63, 0x44,
0x59, 0x96, 0x2a, 0x82, 0xb2, 0x5e, 0xb8, 0x82, 0xb4, 0x28, 0xb8, 0x3b, 0xe2, 0xde, 0x4f, 0x37,
0xd3, 0x35, 0x38, 0xcd, 0x64, 0x33, 0x53, 0xb5, 0xb7, 0xbe, 0x8b, 0x4f, 0xe4, 0x0b, 0xc9, 0x4c,
0x7e, 0x9a, 0x2c, 0x74, 0xef, 0x72, 0xbe, 0xef, 0x3b, 0x67, 0xce, 0x6f, 0x60, 0x94, 0xa7, 0x42,
0x2a, 0x76, 0x94, 0x17, 0x52, 0x4b, 0xf4, 0x4b, 0x2b, 0x7a, 0x03, 0x77, 0xce, 0x37, 0xbc, 0xd8,
0x9e, 0x9d, 0x7f, 0xa6, 0xfc, 0x7a, 0xc3, 0x95, 0xc6, 0x07, 0x70, 0x90, 0x66, 0x09, 0xff, 0x3d,
0x21, 0x21, 0x89, 0x03, 0x5a, 0x1a, 0x78, 0x17, 0xdc, 0xfc, 0x5a, 0x4c, 0x7a, 0x16, 0x33, 0x9f,
0xd1, 0x1a, 0x86, 0x54, 0xfe, 0xa2, 0x5c, 0xe5, 0x32, 0x53, 0x1c, 0x5f, 0x40, 0xff, 0x3b, 0x67,
0x09, 0x2f, 0xd4, 0x84, 0x84, 0x6e, 0x3c, 0x9c, 0xe1, 0x51, 0xf5, 0xe2, 0x07, 0x29, 0x36, 0xeb,
0x6c, 0x91, 0xad, 0x24, 0xad, 0x25, 0x78, 0x0c, 0xfd, 0x4b, 0x0b, 0xab, 0x49, 0xcf, 0xaa, 0x0f,
0xbb, 0xea, 0x3a, 0x2c, 0xad, 0x65, 0xd1, 0x09, 0xc0, 0x2e, 0x10, 0x22, 0x78, 0x19, 0x5b, 0xf3,
0x2a, 0x47, 0xfb, 0x8d, 0x53, 0x18, 0x24, 0x4c, 0x33, 0xbd, 0xcd, 0x79, 0x95, 0x67, 0x63, 0x47,
0xff, 0x7a, 0x30, 0xee, 0x46, 0xc6, 0xc7, 0x10, 0x28, 0x5d, 0xa4, 0xd9, 0xd5, 0x05, 0x13, 0x65,
0x9c, 0xb9, 0x43, 0x77, 0x90, 0xe1, 0x37, 0x69, 0xa6, 0x5f, 0xbf, 0x32, 0xbc, 0x89, 0xe7, 0x19,
0xbe, 0x81, 0xf0, 0x11, 0x0c, 0x1a, 0xda, 0x0d, 0x49, 0xec, 0xce, 0x1d, 0xda, 0x20, 0x38, 0x85,
0xfe, 0x52, 0x4a, 0x61, 0x48, 0x2f, 0x24, 0xf1, 0x60, 0xee, 0xd0, 0x1a, 0xb0, 0x9c, 0x90, 0x4b,
0xc3, 0x1d, 0x84, 0x24, 0x1e, 0x59, 0xae, 0x04, 0xf0, 0x1d, 0x8c, 0xcb, 0x27, 0xde, 0x17, 0x05,
0xdb, 0x1a, 0x89, 0x1f, 0x92, 0x78, 0x38, 0xbb, 0x5f, 0xf7, 0xe7, 0xdb, 0x8e, 0x9d, 0x3b, 0xf4,
0x86, 0xd8, 0xb8, 0x97, 0x15, 0x34, 0xee, 0xfd, 0xae, 0xfb, 0xd7, 0x1d, 0x6b, 0xdc, 0xbb, 0x62,
0x0c, 0x01, 0x56, 0x42, 0xb2, 0xaa, 0xaa, 0x41, 0x48, 0x62, 0x32, 0x77, 0x68, 0x0b, 0x3b, 0x1d,
0x42, 0x50, 0x4e, 0xe4, 0x82, 0x89, 0xe8, 0x2f, 0x81, 0xf1, 0x22, 0x53, 0x39, 0xbf, 0xd4, 0xb7,
0x6f, 0xcf, 0xf3, 0xf6, 0xb8, 0x4d, 0x3e, 0xf7, 0xea, 0x7c, 0x16, 0x89, 0xfa, 0x52, 0x7c, 0xe2,
0x5b, 0xd5, 0x4c, 0x1a, 0x23, 0x18, 0xad, 0x52, 0xa1, 0x79, 0xf1, 0x31, 0xe5, 0x22, 0x51, 0x13,
0x37, 0x74, 0xe3, 0x80, 0x76, 0x30, 0xf3, 0x8c, 0x48, 0xd7, 0xa9, 0xb6, 0xcd, 0xf5, 0x68, 0x69,
0xe0, 0x21, 0xf8, 0x72, 0xb5, 0x52, 0x5c, 0xdb, 0xbe, 0x7a, 0xb4, 0xb2, 0xa2, 0x27, 0x30, 0x6c,
0xb5, 0xcd, 0x2c, 0xcf, 0x4f, 0x26, 0xca, 0x3d, 0xf5, 0xa8, 0xfd, 0x36, 0x92, 0x56, 0x6b, 0x3a,
0x92, 0xa0, 0x92, 0x5c, 0x41, 0xd0, 0x64, 0x8b, 0x4f, 0xc1, 0x4d, 0x13, 0x65, 0xab, 0xdc, 0x3b,
0x1c, 0xa3, 0xc0, 0x67, 0xe0, 0xfd, 0xe0, 0xdb, 0xba, 0xee, 0x3d, 0x73, 0xb0, 0x92, 0x53, 0x1f,
0x3c, 0xb3, 0xac, 0xb3, 0x3f, 0x04, 0xfc, 0x33, 0x2b, 0xc3, 0x13, 0x18, 0xd4, 0xf7, 0x89, 0x0f,
0x6b, 0xdf, 0x1b, 0x17, 0x3b, 0x6d, 0x82, 0xb6, 0xee, 0x31, 0x72, 0x8e, 0x09, 0xbe, 0x85, 0x7e,
0x35, 0x1e, 0x6c, 0xee, 0xab, 0x3b, 0xaf, 0xbd, 0xbe, 0x4b, 0xdf, 0xfe, 0x28, 0x5e, 0xfe, 0x0f,
0x00, 0x00, 0xff, 0xff, 0x0a, 0x57, 0xf8, 0xfb, 0x38, 0x04, 0x00, 0x00,
}

2
proto/pilosa.proto
View file

@ -34,6 +34,8 @@ message InspectRequest {
string index = 1;
IdsOrKeys columns = 2;
repeated string filterFields = 3;
uint64 limit = 4;
uint64 offset = 5;
}
message Uint64Array {

2
row.go
View file

@ -18,7 +18,7 @@ import (
"encoding/json"
"sort"
"github.com/pilosa/pilosa/v2/ext"
"github.com/molecula/ext"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pkg/errors"
)

78
server.go
View file

@ -17,19 +17,19 @@ package pilosa
import (
"context"
"fmt"
"io"
"log"
"os"
"os/exec"
"path/filepath"
"plugin"
"runtime"
"strconv"
"strings"
"sync"
"time"
"github.com/pilosa/pilosa/v2/ext"
"github.com/molecula/ext"
// extensions pulls in some extensions depending on build tags
_ "github.com/pilosa/pilosa/v2/extensions"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
@ -61,7 +61,6 @@ type Server struct { // nolint: maligned
hosts []string
clusterDisabled bool
serializer Serializer
extensionPath string
extensions []*ext.ExtensionInfo
// External
@ -341,8 +340,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
if err != nil {
return nil, err
}
s.extensionPath = filepath.Join(path, ".extensions")
s.holder.Path = path
// s.holder.translateFile.Path = filepath.Join(path, ".keys")
s.holder.Logger = s.logger
@ -383,7 +380,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.broadcaster = s
s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
s.holder.broadcaster = s
err = s.loadPlugins()
err = s.loadExtensions()
if err != nil {
s.logger.Printf("not all plugins loaded successfully")
}
@ -420,67 +417,20 @@ func (s *Server) InternalClient() InternalClient {
return s.defaultClient
}
func (s *Server) loadPlugins() error {
var anyError error
dir, err := os.Open(s.extensionPath)
if err != nil {
// don't complain about it not existing, that's fine.
if os.IsNotExist(err) {
s.logger.Printf("extension interface v0: no extensions directory.")
return nil
}
return errors.Wrap(err, "opening extension path:")
}
defer dir.Close()
for files, err := dir.Readdir(64); err != io.EOF; files, err = dir.Readdir(64) {
if err != nil {
return errors.Wrap(err, "searching extension directory:")
}
for _, file := range files {
name := file.Name()
// only .so files are likely plugins.
if !strings.HasSuffix(name, ".so") {
continue
}
// only regular files are candidates for loading.
mode := file.Mode()
if !mode.IsRegular() {
s.logger.Printf("extension file '%s' is not a regular file", name)
continue
}
err = s.loadPlugin(name)
if err != nil {
s.logger.Printf("loading extension %s: %v", name, err)
anyError = err
}
func (s *Server) loadExtensions() error {
exts := ext.NewExtensions()
var lastError error
for _, extension := range exts {
if err := s.loadExtension(extension); err != nil {
lastError = err
}
}
return anyError
return lastError
}
func (s *Server) loadPlugin(name string) error {
path := filepath.Join(s.extensionPath, name)
p, err := plugin.Open(path)
if err != nil {
return err
}
pluginExtInfo, err := p.Lookup("ExtensionInfo")
if err != nil {
return fmt.Errorf("%s: no ExtensionInfo found", name)
}
extInfoFunc, ok := pluginExtInfo.(func(string) (*ext.ExtensionInfo, error))
if !ok {
return fmt.Errorf("%s: unexpected %T instead of ExtensionInfo object", name, pluginExtInfo)
}
extInfo, err := extInfoFunc("v0")
if err != nil {
return errors.Wrap(err, name)
}
if extInfo == nil {
return fmt.Errorf("%s: nil ExtensionInfo", name)
}
func (s *Server) loadExtension(extInfo *ext.ExtensionInfo) error {
if extInfo.ExtensionAPI != "v0" {
return fmt.Errorf("%s: unsupported extension API %s", name, extInfo.ExtensionAPI)
return fmt.Errorf("%s: unsupported extension API %s", extInfo.Name, extInfo.ExtensionAPI)
}
s.extensions = append(s.extensions, extInfo)
bitmapOps := extInfo.BitmapOps
@ -500,7 +450,7 @@ func (s *Server) loadPlugin(name string) error {
unknownOps++
}
}
err = s.executor.registerOps(bitmapOps)
err := s.executor.registerOps(bitmapOps)
if err != nil {
s.logger.Printf("warning: extension registration failed: %v", err)
} else {

106
server/grpc.go
View file

@ -71,6 +71,8 @@ func (h grpcHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQL
// 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 errors.Wrap(err, "getting index")
@ -96,7 +98,17 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
return nil
}
if ints, ok := req.Columns.Type.(*pb.IdsOrKeys_Ids); ok {
limit := req.Limit
if limit == 0 {
limit = defaultLimit
}
offset := req.Offset
if !index.Options().Keys {
ints, ok := req.Columns.Type.(*pb.IdsOrKeys_Ids)
if !ok {
return errors.New("invalid int columns")
}
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "uint64"},
}
@ -104,7 +116,46 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: field.Type()}) // TODO: field.Type likely doesn't align with supported datatypes
}
for _, col := range ints.Ids.Vals {
// 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{
@ -178,6 +229,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "decimal":
value, exists, err := field.FloatValue(col)
if err != nil {
@ -189,6 +241,7 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "bool":
pql := fmt.Sprintf("Rows(%s, column=%d)", field.Name(), col)
query := pilosa.QueryRequest{
@ -216,6 +269,7 @@ 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})
@ -227,7 +281,11 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
}
}
} else if keys, ok := req.Columns.Type.(*pb.IdsOrKeys_Keys); ok {
} else {
keys, ok := req.Columns.Type.(*pb.IdsOrKeys_Keys)
if !ok {
return errToStatusError(errors.New("invalid key columns"))
}
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "string"},
}
@ -235,7 +293,47 @@ func (h grpcHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSer
ci = append(ci, &pb.ColumnInfo{Name: field.Name(), Datatype: field.Type()}) // TODO: field.Type likely doesn't align with supported datatypes
}
for _, col := range keys.Keys.Vals {
// If Columns is empty, then get the _exists list (via All()),
// from the index and loop over that instead.
cols := keys.Keys.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 {
fmt.Println("GOT ERROR trying to get ALL():", err)
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{