From 361e51cb41ce83bc56af744a5c08e432e61bc2cb Mon Sep 17 00:00:00 2001 From: Travis Date: Mon, 16 Dec 2019 18:35:55 -0600 Subject: [PATCH 1/6] Add All() support to PQL, including limit and offset This PR is meant to get all columns from an index based on the TrackExistence row. `All()` is a PQL function that can be used as a typical row object. Optional arguments are `limit` and `offset`. --- api/client/grpc.go | 5 +- executor.go | 138 +++++++++++++++++- executor_test.go | 143 +++++++++++++++++++ go.mod | 4 +- go.sum | 4 + pql/ast.go | 15 +- proto/pilosa.pb.go | 343 +++++++++++++++++++++++++++++++++++++-------- proto/pilosa.proto | 2 + server/grpc.go | 106 +++++++++++++- 9 files changed, 689 insertions(+), 71 deletions(-) diff --git a/api/client/grpc.go b/api/client/grpc.go index 2d99ea76e..fa9f33b68 100644 --- a/api/client/grpc.go +++ b/api/client/grpc.go @@ -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 { diff --git a/executor.go b/executor.go index 25dfbf4c1..47c08043b 100644 --- a/executor.go +++ b/executor.go @@ -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") diff --git a/executor_test.go b/executor_test.go index 665ef9bba..3750a04ad 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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 diff --git a/go.mod b/go.mod index eb38d9288..f47f54dc1 100644 --- a/go.mod +++ b/go.mod @@ -34,15 +34,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 diff --git a/go.sum b/go.sum index 6607cf98d..177cd241f 100644 --- a/go.sum +++ b/go.sum @@ -153,6 +153,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 +212,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= diff --git a/pql/ast.go b/pql/ast.go index 7749c02dd..61b4b6540 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -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{}{ diff --git a/proto/pilosa.pb.go b/proto/pilosa.pb.go index b81722e06..1a5bfca49 100644 --- a/proto/pilosa.pb.go +++ b/proto/pilosa.pb.go @@ -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, +} diff --git a/proto/pilosa.proto b/proto/pilosa.proto index a8822ec46..9a5828624 100644 --- a/proto/pilosa.proto +++ b/proto/pilosa.proto @@ -34,6 +34,8 @@ message InspectRequest { string index = 1; IdsOrKeys columns = 2; repeated string filterFields = 3; + uint64 limit = 4; + uint64 offset = 5; } message Uint64Array { diff --git a/server/grpc.go b/server/grpc.go index cb1856b9f..a0ef90cbe 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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{ From f51c2dbc42628d73ea2fc0aed52941fc78515f08 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 2 Dec 2019 13:24:53 -0600 Subject: [PATCH 2/6] use extensions through build tags --- Makefile | 1 + ext/ext.go | 25 +++++++++++++ ext/extensions/distinct.go | 21 +++++++++++ ext/extensions/dummy.go | 18 +++++++++ server.go | 75 +++++++------------------------------- 5 files changed, 78 insertions(+), 62 deletions(-) create mode 100644 ext/extensions/distinct.go create mode 100644 ext/extensions/dummy.go diff --git a/Makefile b/Makefile index bee802120..ac204d3ca 100644 --- a/Makefile +++ b/Makefile @@ -16,6 +16,7 @@ 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 diff --git a/ext/ext.go b/ext/ext.go index b6daa3ef0..762ae44f8 100644 --- a/ext/ext.go +++ b/ext/ext.go @@ -38,6 +38,8 @@ // case, no ops are registered. package ext +import "sync" + // The Bitmap type represents a Pilosa bitmap, and is used for bitmap // operations. type Bitmap interface { @@ -236,3 +238,26 @@ type ExtensionInfo struct { License string // License info. BitmapOps []BitmapOp // List of provided ops. } + +var extMu sync.Mutex + +var knownExtensions []*ExtensionInfo +var newExtensions []*ExtensionInfo + +// RegisterExtension tells the extension system about a new extension. +func RegisterExtension(ext *ExtensionInfo) { + extMu.Lock() + defer extMu.Unlock() + newExtensions = append(newExtensions, ext) +} + +// NewExtentsions returns extensions that have been registered, but not previously +// returned by Newextensions. +func NewExtensions() []*ExtensionInfo { + extMu.Lock() + defer extMu.Unlock() + knownExtensions = append(knownExtensions, newExtensions...) + ret := newExtensions + newExtensions = nil + return ret +} diff --git a/ext/extensions/distinct.go b/ext/extensions/distinct.go new file mode 100644 index 000000000..42f11c90f --- /dev/null +++ b/ext/extensions/distinct.go @@ -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" +) diff --git a/ext/extensions/dummy.go b/ext/extensions/dummy.go new file mode 100644 index 000000000..cf418edb5 --- /dev/null +++ b/ext/extensions/dummy.go @@ -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 diff --git a/server.go b/server.go index ddf43ed50..645418898 100644 --- a/server.go +++ b/server.go @@ -17,12 +17,10 @@ package pilosa import ( "context" "fmt" - "io" "log" "os" "os/exec" "path/filepath" - "plugin" "runtime" "strconv" "strings" @@ -30,6 +28,8 @@ import ( "time" "github.com/pilosa/pilosa/v2/ext" + // extensions pulls in some extensions depending on build tags + _ "github.com/pilosa/pilosa/v2/ext/extensions" "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/roaring" @@ -341,8 +341,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 +381,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 +418,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 +451,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 { From 0eba050054aa260cb780d3dab7f67c5f89c77c5a Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 2 Dec 2019 13:42:51 -0600 Subject: [PATCH 3/6] stop using pkg/plugin, start using build tags After a few experiments with pkg/plugin, I'm ready to concede that the people warning me it was unsuitable for production use were in fact correct. In the brave new world, the "ext" package is moved to its own module outside pilosa. This means that importing it doesn't imply any need to version-check against pilosa; we can just use versioned copies of the ext package, which can be public because it doesn't contain anything we need to care about keeping proprietary. Then we can, conditional on build tags, import modules from a neighboring repo which contains the actual implementations, and if they're imported, their init functions register them. --- executor.go | 2 +- ext/ext.go | 263 --------------------- ext/samples/.gitignore | 1 - ext/samples/some/some.go | 125 ---------- extension.go | 2 +- {ext/extensions => extensions}/distinct.go | 0 {ext/extensions => extensions}/dummy.go | 0 pql/ast.go | 2 +- row.go | 2 +- server.go | 4 +- 10 files changed, 6 insertions(+), 395 deletions(-) delete mode 100644 ext/ext.go delete mode 100644 ext/samples/.gitignore delete mode 100644 ext/samples/some/some.go rename {ext/extensions => extensions}/distinct.go (100%) rename {ext/extensions => extensions}/dummy.go (100%) diff --git a/executor.go b/executor.go index 47c08043b..f9586f958 100644 --- a/executor.go +++ b/executor.go @@ -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" diff --git a/ext/ext.go b/ext/ext.go deleted file mode 100644 index 762ae44f8..000000000 --- a/ext/ext.go +++ /dev/null @@ -1,263 +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 - -import "sync" - -// 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. -} - -var extMu sync.Mutex - -var knownExtensions []*ExtensionInfo -var newExtensions []*ExtensionInfo - -// RegisterExtension tells the extension system about a new extension. -func RegisterExtension(ext *ExtensionInfo) { - extMu.Lock() - defer extMu.Unlock() - newExtensions = append(newExtensions, ext) -} - -// NewExtentsions returns extensions that have been registered, but not previously -// returned by Newextensions. -func NewExtensions() []*ExtensionInfo { - extMu.Lock() - defer extMu.Unlock() - knownExtensions = append(knownExtensions, newExtensions...) - ret := newExtensions - newExtensions = nil - return ret -} diff --git a/ext/samples/.gitignore b/ext/samples/.gitignore deleted file mode 100644 index a63fa2c94..000000000 --- a/ext/samples/.gitignore +++ /dev/null @@ -1 +0,0 @@ -*/*.so diff --git a/ext/samples/some/some.go b/ext/samples/some/some.go deleted file mode 100644 index 07088d255..000000000 --- a/ext/samples/some/some.go +++ /dev/null @@ -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") -} diff --git a/extension.go b/extension.go index d449f3f0c..1a492cce3 100644 --- a/extension.go +++ b/extension.go @@ -17,7 +17,7 @@ package pilosa import ( "fmt" - "github.com/pilosa/pilosa/v2/ext" + "github.com/molecula/ext" "github.com/pilosa/pilosa/v2/roaring" ) diff --git a/ext/extensions/distinct.go b/extensions/distinct.go similarity index 100% rename from ext/extensions/distinct.go rename to extensions/distinct.go diff --git a/ext/extensions/dummy.go b/extensions/dummy.go similarity index 100% rename from ext/extensions/dummy.go rename to extensions/dummy.go diff --git a/pql/ast.go b/pql/ast.go index 61b4b6540..738601100 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -23,7 +23,7 @@ import ( "strings" "time" - "github.com/pilosa/pilosa/v2/ext" + "github.com/molecula/ext" ) // Query represents a PQL query. diff --git a/row.go b/row.go index d9a4f9cc9..82a67cf8f 100644 --- a/row.go +++ b/row.go @@ -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" ) diff --git a/server.go b/server.go index 645418898..44e27b1a5 100644 --- a/server.go +++ b/server.go @@ -27,9 +27,9 @@ import ( "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/ext/extensions" + _ "github.com/pilosa/pilosa/v2/extensions" "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/roaring" From 7e1fd8392fa8cf8bd530b00170c88d4e3e84dc33 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 2 Dec 2019 13:58:36 -0600 Subject: [PATCH 4/6] go.mod/go.sum changes for using molecula/ext This pins us to the initial external release of molecula/ext, which with any luck will be the only one. (Narrator: It was not to be the only one.) We also use GOPRIVATE so we don't need a replace directive. --- Makefile | 1 + go.mod | 3 ++- go.sum | 4 ++++ server.go | 1 - 4 files changed, 7 insertions(+), 2 deletions(-) diff --git a/Makefile b/Makefile index ac204d3ca..55356bd59 100644 --- a/Makefile +++ b/Makefile @@ -23,6 +23,7 @@ endef LICENSE_HASH=$(shell $(call LICENSE_HASH_CODE, pilosa.go)) export GO111MODULE=on +export GOPRIVATE=github.com/molecula # Run tests and compile Pilosa default: test build diff --git a/go.mod b/go.mod index f47f54dc1..9366e18cd 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 177cd241f..1b1b7af2f 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/server.go b/server.go index 44e27b1a5..96a4d5cde 100644 --- a/server.go +++ b/server.go @@ -61,7 +61,6 @@ type Server struct { // nolint: maligned hosts []string clusterDisabled bool serializer Serializer - extensionPath string extensions []*ext.ExtensionInfo // External From 49b2029656d882f3a72b7d947ae9f67f965f063a Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Thu, 19 Dec 2019 10:24:15 -0600 Subject: [PATCH 5/6] Run "go mod vendor" outside of Docker so authenticated modules may use system credentials --- Dockerfile-clustertests | 5 +---- Makefile | 4 ++-- internal/clustertests/docker-compose.yml | 2 +- 3 files changed, 4 insertions(+), 7 deletions(-) diff --git a/Dockerfile-clustertests b/Dockerfile-clustertests index 69fbd9a54..6241ae70d 100644 --- a/Dockerfile-clustertests +++ b/Dockerfile-clustertests @@ -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 diff --git a/Makefile b/Makefile index 55356bd59..20f00ac03 100644 --- a/Makefile +++ b/Makefile @@ -87,14 +87,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 diff --git a/internal/clustertests/docker-compose.yml b/internal/clustertests/docker-compose.yml index 1546a5b13..36b418921 100644 --- a/internal/clustertests/docker-compose.yml +++ b/internal/clustertests/docker-compose.yml @@ -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: From fe0f57651ed24e4b2e30b0b491ef14fa411475a0 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 20 Dec 2019 12:19:57 -0600 Subject: [PATCH 6/6] build with distinct by default --- Makefile | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Makefile b/Makefile index 20f00ac03..9d24ae75f 100644 --- a/Makefile +++ b/Makefile @@ -22,8 +22,10 @@ define LICENSE_HASH_CODE 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