mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 02:44:59 +00:00
78ff75690d
pass previous+1 directly to fragment.rows so that the iterator can seek directly to the start point. handle limit inside reduce so it can skip out early and avoid extra allocation.
2297 lines
61 KiB
Go
2297 lines
61 KiB
Go
// Copyright 2017 Pilosa Corp.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package pilosa
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/pilosa/pilosa/pql"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
// defaultField is the field used if one is not specified.
|
|
const (
|
|
defaultField = "general"
|
|
|
|
// defaultMinThreshold is the lowest count to use in a Top-N operation when
|
|
// looking for additional id/count pairs.
|
|
defaultMinThreshold = 1
|
|
|
|
columnLabel = "col"
|
|
rowLabel = "row"
|
|
)
|
|
|
|
// executor recursively executes calls in a PQL query across all shards.
|
|
type executor struct {
|
|
Holder *Holder
|
|
|
|
// Local hostname & cluster configuration.
|
|
Node *Node
|
|
Cluster *cluster
|
|
|
|
// Client used for remote requests.
|
|
client InternalQueryClient
|
|
|
|
// Maximum number of Set() or Clear() commands per request.
|
|
MaxWritesPerRequest int
|
|
|
|
// Stores key/id translation data.
|
|
TranslateStore TranslateStore
|
|
}
|
|
|
|
// executorOption is a functional option type for pilosa.Executor
|
|
type executorOption func(e *executor) error
|
|
|
|
func optExecutorInternalQueryClient(c InternalQueryClient) executorOption {
|
|
return func(e *executor) error {
|
|
e.client = c
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// newExecutor returns a new instance of Executor.
|
|
func newExecutor(opts ...executorOption) *executor {
|
|
e := &executor{
|
|
client: newNopInternalQueryClient(),
|
|
}
|
|
for _, opt := range opts {
|
|
err := opt(e)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
}
|
|
return e
|
|
}
|
|
|
|
// Execute executes a PQL query.
|
|
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) ([]interface{}, error) {
|
|
// Verify that an index is set.
|
|
if index == "" {
|
|
return nil, ErrIndexRequired
|
|
}
|
|
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
|
|
// Verify that the number of writes do not exceed the maximum.
|
|
if e.MaxWritesPerRequest > 0 && q.WriteCallN() > e.MaxWritesPerRequest {
|
|
return nil, ErrTooManyWrites
|
|
}
|
|
|
|
// Default options.
|
|
if opt == nil {
|
|
opt = &execOptions{}
|
|
}
|
|
|
|
// Translate query keys to ids, if necessary.
|
|
// No need to translate a remote call.
|
|
if !opt.Remote {
|
|
for i := range q.Calls {
|
|
if err := e.translateCall(index, idx, q.Calls[i]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
|
|
results, err := e.execute(ctx, index, q, shards, opt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Translate response objects from ids to keys, if necessary.
|
|
// No need to translate a remote call.
|
|
if !opt.Remote {
|
|
for i := range results {
|
|
results[i], err = e.translateResult(index, idx, q.Calls[i], results[i])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) ([]interface{}, error) {
|
|
// Don't bother calculating shards for query types that don't require it.
|
|
needsShards := needsShards(q.Calls)
|
|
|
|
// If shards are specified, then use that value for shards. If shards aren't
|
|
// specified, then include all of them.
|
|
if len(shards) == 0 && needsShards {
|
|
// Round up the number of shards.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
shards = idx.AvailableShards().Slice()
|
|
if len(shards) == 0 {
|
|
shards = []uint64{0}
|
|
}
|
|
}
|
|
|
|
// Optimize handling for bulk attribute insertion.
|
|
if hasOnlySetRowAttrs(q.Calls) {
|
|
return e.executeBulkSetRowAttrs(ctx, index, q.Calls, opt)
|
|
}
|
|
|
|
// Execute each call serially.
|
|
results := make([]interface{}, 0, len(q.Calls))
|
|
for _, call := range q.Calls {
|
|
v, err := e.executeCall(ctx, index, call, shards, opt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
results = append(results, v)
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
// executeCall executes a call.
|
|
func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
|
|
if err := e.validateCallArgs(c); err != nil {
|
|
return nil, errors.Wrap(err, "validating args")
|
|
}
|
|
indexTag := fmt.Sprintf("index:%s", index)
|
|
// Special handling for mutation and top-n calls.
|
|
switch c.Name {
|
|
case "Sum":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeSum(ctx, index, c, shards, opt)
|
|
case "Min":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeMin(ctx, index, c, shards, opt)
|
|
case "Max":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeMax(ctx, index, c, shards, opt)
|
|
case "Clear":
|
|
return e.executeClearBit(ctx, index, c, opt)
|
|
case "Count":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeCount(ctx, index, c, shards, opt)
|
|
case "Set":
|
|
return e.executeSet(ctx, index, c, opt)
|
|
case "SetRowAttrs":
|
|
return nil, e.executeSetRowAttrs(ctx, index, c, opt)
|
|
case "SetColumnAttrs":
|
|
return nil, e.executeSetColumnAttrs(ctx, index, c, opt)
|
|
case "TopN":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeTopN(ctx, index, c, shards, opt)
|
|
case "Rows":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeRows(ctx, index, c, shards, opt)
|
|
case "GroupBy":
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeGroupBy(ctx, index, c, shards, opt)
|
|
case "Options":
|
|
return e.executeOptionsCall(ctx, index, c, shards, opt)
|
|
default:
|
|
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
|
return e.executeBitmapCall(ctx, index, c, shards, opt)
|
|
}
|
|
}
|
|
|
|
// validateCallArgs ensures that the value types in call.Args are expected.
|
|
func (e *executor) validateCallArgs(c *pql.Call) error {
|
|
if _, ok := c.Args["ids"]; ok {
|
|
switch v := c.Args["ids"].(type) {
|
|
case []int64, []uint64:
|
|
// noop
|
|
case []interface{}:
|
|
b := make([]int64, len(v))
|
|
for i := range v {
|
|
b[i] = v[i].(int64)
|
|
}
|
|
c.Args["ids"] = b
|
|
default:
|
|
return fmt.Errorf("invalid call.Args[ids]: %s", v)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *executor) executeOptionsCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
|
|
optCopy := &execOptions{}
|
|
*optCopy = *opt
|
|
if arg, ok := c.Args["columnAttrs"]; ok {
|
|
if value, ok := arg.(bool); ok {
|
|
opt.ColumnAttrs = value
|
|
} else {
|
|
return nil, errors.New("Query(): columnAttrs must be a bool")
|
|
}
|
|
}
|
|
if arg, ok := c.Args["excludeRowAttrs"]; ok {
|
|
if value, ok := arg.(bool); ok {
|
|
optCopy.ExcludeRowAttrs = value
|
|
} else {
|
|
return nil, errors.New("Query(): excludeRowAttrs must be a bool")
|
|
}
|
|
}
|
|
if arg, ok := c.Args["excludeColumns"]; ok {
|
|
if value, ok := arg.(bool); ok {
|
|
optCopy.ExcludeColumns = value
|
|
} else {
|
|
return nil, errors.New("Query(): excludeColumns must be a bool")
|
|
}
|
|
}
|
|
if arg, ok := c.Args["shards"]; ok {
|
|
if optShards, ok := arg.([]interface{}); ok {
|
|
shards = []uint64{}
|
|
for _, s := range optShards {
|
|
if shard, ok := s.(int64); ok {
|
|
shards = append(shards, uint64(shard))
|
|
} else {
|
|
return nil, errors.New("Query(): shards must be a list of unsigned integers")
|
|
}
|
|
|
|
}
|
|
} else {
|
|
return nil, errors.New("Query(): shards must be a list of unsigned integers")
|
|
}
|
|
}
|
|
return e.executeCall(ctx, index, c.Children[0], shards, optCopy)
|
|
}
|
|
|
|
// executeSum executes a Sum() call.
|
|
func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
|
|
if field := c.Args["field"]; field == "" {
|
|
return ValCount{}, errors.New("Sum(): field required")
|
|
}
|
|
|
|
if len(c.Children) > 1 {
|
|
return ValCount{}, errors.New("Sum() only accepts a single bitmap input")
|
|
}
|
|
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeSumCountShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.(ValCount)
|
|
return other.add(v.(ValCount))
|
|
}
|
|
|
|
result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
other, _ := result.(ValCount)
|
|
|
|
if other.Count == 0 {
|
|
return ValCount{}, nil
|
|
}
|
|
return other, nil
|
|
}
|
|
|
|
// executeMin executes a Min() call.
|
|
func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
|
|
if field := c.Args["field"]; field == "" {
|
|
return ValCount{}, errors.New("Min(): field required")
|
|
}
|
|
|
|
if len(c.Children) > 1 {
|
|
return ValCount{}, errors.New("Min() only accepts a single bitmap input")
|
|
}
|
|
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeMinShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.(ValCount)
|
|
return other.smaller(v.(ValCount))
|
|
}
|
|
|
|
result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
other, _ := result.(ValCount)
|
|
|
|
if other.Count == 0 {
|
|
return ValCount{}, nil
|
|
}
|
|
return other, nil
|
|
}
|
|
|
|
// executeMax executes a Max() call.
|
|
func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
|
|
if field := c.Args["field"]; field == "" {
|
|
return ValCount{}, errors.New("Max(): field required")
|
|
}
|
|
|
|
if len(c.Children) > 1 {
|
|
return ValCount{}, errors.New("Max() only accepts a single bitmap input")
|
|
}
|
|
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeMaxShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.(ValCount)
|
|
return other.larger(v.(ValCount))
|
|
}
|
|
|
|
result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
other, _ := result.(ValCount)
|
|
|
|
if other.Count == 0 {
|
|
return ValCount{}, nil
|
|
}
|
|
return other, nil
|
|
}
|
|
|
|
// executeBitmapCall executes a call that returns a bitmap.
|
|
func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeBitmapCallShard(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
|
|
}
|
|
|
|
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "map reduce")
|
|
}
|
|
|
|
// Attach attributes for Row() calls.
|
|
// If the column label is used then return column attributes.
|
|
// If the row label is used then return bitmap attributes.
|
|
row, _ := other.(*Row)
|
|
if c.Name == "Row" {
|
|
if opt.ExcludeRowAttrs {
|
|
row.Attrs = map[string]interface{}{}
|
|
} else {
|
|
idx := e.Holder.Index(index)
|
|
if idx != nil {
|
|
if columnID, ok, err := c.UintArg("_" + columnLabel); ok && err == nil {
|
|
attrs, err := idx.ColumnAttrStore().Attrs(columnID)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting column attrs")
|
|
}
|
|
row.Attrs = attrs
|
|
} else if err != nil {
|
|
return nil, err
|
|
} else {
|
|
// field, _ := c.Args["field"].(string)
|
|
fieldName, _ := c.FieldArg()
|
|
if fr := idx.Field(fieldName); fr != nil {
|
|
rowID, _, err := c.UintArg(fieldName)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting row")
|
|
}
|
|
attrs, err := fr.RowAttrStore().Attrs(rowID)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting row attrs")
|
|
}
|
|
row.Attrs = attrs
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if opt.ExcludeColumns {
|
|
row.segments = []rowSegment{}
|
|
}
|
|
|
|
return row, nil
|
|
}
|
|
|
|
// executeBitmapCallShard executes a bitmap call for a single shard.
|
|
func (e *executor) executeBitmapCallShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
switch c.Name {
|
|
case "Row":
|
|
return e.executeBitmapShard(ctx, index, c, shard)
|
|
case "Difference":
|
|
return e.executeDifferenceShard(ctx, index, c, shard)
|
|
case "Intersect":
|
|
return e.executeIntersectShard(ctx, index, c, shard)
|
|
case "Range":
|
|
return e.executeRangeShard(ctx, index, c, shard)
|
|
case "Union":
|
|
return e.executeUnionShard(ctx, index, c, shard)
|
|
case "Xor":
|
|
return e.executeXorShard(ctx, index, c, shard)
|
|
case "Not":
|
|
return e.executeNotShard(ctx, index, c, shard)
|
|
default:
|
|
return nil, fmt.Errorf("unknown call: %s", c.Name)
|
|
}
|
|
}
|
|
|
|
// executeSumCountShard calculates the sum and count for bsiGroups on a shard.
|
|
func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
|
|
var filter *Row
|
|
if len(c.Children) == 1 {
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return ValCount{}, errors.Wrap(err, "executing bitmap call")
|
|
}
|
|
filter = row
|
|
}
|
|
|
|
fieldName, _ := c.Args["field"].(string)
|
|
|
|
field := e.Holder.Field(index, fieldName)
|
|
if field == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
bsig := field.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if fragment == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
vsum, vcount, err := fragment.sum(filter, bsig.BitDepth())
|
|
if err != nil {
|
|
return ValCount{}, errors.Wrap(err, "computing sum")
|
|
}
|
|
return ValCount{
|
|
Val: int64(vsum) + (int64(vcount) * bsig.Min),
|
|
Count: int64(vcount),
|
|
}, nil
|
|
}
|
|
|
|
// executeMinShard calculates the min for bsiGroups on a shard.
|
|
func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
|
|
var filter *Row
|
|
if len(c.Children) == 1 {
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
filter = row
|
|
}
|
|
|
|
fieldName, _ := c.Args["field"].(string)
|
|
|
|
field := e.Holder.Field(index, fieldName)
|
|
if field == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
bsig := field.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if fragment == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fmin, fcount, err := fragment.min(filter, bsig.BitDepth())
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
return ValCount{
|
|
Val: int64(fmin) + bsig.Min,
|
|
Count: int64(fcount),
|
|
}, nil
|
|
}
|
|
|
|
// executeMaxShard calculates the max for bsiGroups on a shard.
|
|
func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
|
|
var filter *Row
|
|
if len(c.Children) == 1 {
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
filter = row
|
|
}
|
|
|
|
fieldName, _ := c.Args["field"].(string)
|
|
|
|
field := e.Holder.Field(index, fieldName)
|
|
if field == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
bsig := field.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if fragment == nil {
|
|
return ValCount{}, nil
|
|
}
|
|
|
|
fmax, fcount, err := fragment.max(filter, bsig.BitDepth())
|
|
if err != nil {
|
|
return ValCount{}, err
|
|
}
|
|
return ValCount{
|
|
Val: int64(fmax) + bsig.Min,
|
|
Count: int64(fcount),
|
|
}, nil
|
|
}
|
|
|
|
// executeTopN executes a TopN() call.
|
|
// This first performs the TopN() to determine the top results and then
|
|
// requeries to retrieve the full counts for each of the top results.
|
|
func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
|
|
idsArg, _, err := c.UintSliceArg("ids")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopN: %v", err)
|
|
}
|
|
n, _, err := c.UintArg("n")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopN: %v", err)
|
|
}
|
|
|
|
// Execute original query.
|
|
pairs, err := e.executeTopNShards(ctx, index, c, shards, opt)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "finding top results")
|
|
}
|
|
|
|
// If this call is against specific ids, or we didn't get results,
|
|
// or we are part of a larger distributed query then don't refetch.
|
|
if len(pairs) == 0 || len(idsArg) > 0 || opt.Remote {
|
|
return pairs, nil
|
|
}
|
|
// Only the original caller should refetch the full counts.
|
|
other := c.Clone()
|
|
|
|
ids := Pairs(pairs).Keys()
|
|
sort.Sort(uint64Slice(ids))
|
|
other.Args["ids"] = ids
|
|
|
|
trimmedList, err := e.executeTopNShards(ctx, index, other, shards, opt)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "retrieving full counts")
|
|
}
|
|
|
|
if n != 0 && int(n) < len(trimmedList) {
|
|
trimmedList = trimmedList[0:n]
|
|
}
|
|
return trimmedList, nil
|
|
}
|
|
|
|
func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeTopNShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.([]Pair)
|
|
return Pairs(other).Add(v.([]Pair))
|
|
}
|
|
|
|
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
results, _ := other.([]Pair)
|
|
|
|
// Sort final merged results.
|
|
sort.Sort(Pairs(results))
|
|
|
|
return results, nil
|
|
}
|
|
|
|
// executeTopNShard executes a TopN call for a single shard.
|
|
func (e *executor) executeTopNShard(ctx context.Context, index string, c *pql.Call, shard uint64) ([]Pair, error) {
|
|
field, _ := c.Args["_field"].(string)
|
|
n, _, err := c.UintArg("n")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopNShard: %v", err)
|
|
}
|
|
attrName, _ := c.Args["attrName"].(string)
|
|
rowIDs, _, err := c.UintSliceArg("ids")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopNShard: %v", err)
|
|
}
|
|
minThreshold, _, err := c.UintArg("threshold")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopNShard: %v", err)
|
|
}
|
|
attrValues, _ := c.Args["attrValues"].([]interface{})
|
|
tanimotoThreshold, _, err := c.UintArg("tanimotoThreshold")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeTopNShard: %v", err)
|
|
}
|
|
|
|
// Retrieve bitmap used to intersect.
|
|
var src *Row
|
|
if len(c.Children) == 1 {
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
src = row
|
|
} else if len(c.Children) > 1 {
|
|
return nil, errors.New("TopN() can only have one input bitmap")
|
|
}
|
|
|
|
// Set default field.
|
|
if field == "" {
|
|
field = defaultField
|
|
}
|
|
|
|
f := e.Holder.fragment(index, field, viewStandard, shard)
|
|
if f == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
if minThreshold <= 0 {
|
|
minThreshold = defaultMinThreshold
|
|
}
|
|
|
|
if tanimotoThreshold > 100 {
|
|
return nil, errors.New("Tanimoto Threshold is from 1 to 100 only")
|
|
}
|
|
return f.top(topOptions{
|
|
N: int(n),
|
|
Src: src,
|
|
RowIDs: rowIDs,
|
|
FilterName: attrName,
|
|
FilterValues: attrValues,
|
|
MinThreshold: minThreshold,
|
|
TanimotoThreshold: tanimotoThreshold,
|
|
})
|
|
}
|
|
|
|
// executeDifferenceShard executes a difference() call for a local shard.
|
|
func (e *executor) executeDifferenceShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
var other *Row
|
|
if len(c.Children) == 0 {
|
|
return nil, fmt.Errorf("empty Difference query is currently not supported")
|
|
}
|
|
for i, input := range c.Children {
|
|
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if i == 0 {
|
|
other = row
|
|
} else {
|
|
other = other.Difference(row)
|
|
}
|
|
}
|
|
other.invalidateCount()
|
|
return other, nil
|
|
}
|
|
|
|
// RowIdentifiers is a return type for a list of
|
|
// row ids or row keys. The names `Rows` and `Keys`
|
|
// are meant to follow the same convention as the
|
|
// Row query which returns `Columns` and `Keys`.
|
|
// TODO: Rename this to something better. Anything.
|
|
type RowIdentifiers struct {
|
|
Rows []uint64 `json:"rows"`
|
|
Keys []string `json:"keys,omitempty"`
|
|
}
|
|
|
|
// RowIDs is a query return type for just uint64 row ids.
|
|
// It should only be used internally (since RowIdentifiers
|
|
// is the external return type), but it is exported because
|
|
// the proto package needs access to it.
|
|
type RowIDs []uint64
|
|
|
|
func (r RowIDs) merge(other RowIDs, limit int) RowIDs {
|
|
i, j := 0, 0
|
|
result := make(RowIDs, 0)
|
|
for i < len(r) && j < len(other) && len(result) < limit {
|
|
av, bv := r[i], other[j]
|
|
if av < bv {
|
|
result = append(result, av)
|
|
i++
|
|
} else if av > bv {
|
|
result = append(result, bv)
|
|
j++
|
|
} else {
|
|
result = append(result, bv)
|
|
i++
|
|
j++
|
|
}
|
|
}
|
|
for i < len(r) && len(result) < limit {
|
|
result = append(result, r[i])
|
|
i++
|
|
}
|
|
for j < len(other) && len(result) < limit {
|
|
result = append(result, other[j])
|
|
j++
|
|
}
|
|
return result
|
|
}
|
|
|
|
func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]GroupCount, error) {
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeGroupByShard(ctx, index, c, shard)
|
|
}
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.([]GroupCount)
|
|
return mergeGroupCounts(other, v.([]GroupCount))
|
|
}
|
|
// Get full result set.
|
|
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
results, _ := other.([]GroupCount)
|
|
|
|
// Apply offset.
|
|
if offset, hasOffset, err := c.UintArg("offset"); err != nil {
|
|
return nil, err
|
|
} else if hasOffset {
|
|
if int(offset) < len(results) {
|
|
results = results[offset:]
|
|
}
|
|
}
|
|
// Apply limit.
|
|
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
|
|
return nil, err
|
|
} else if hasLimit {
|
|
if int(limit) < len(results) {
|
|
results = results[:limit]
|
|
}
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
// FieldRow is used to distinguish rows in a group by result.
|
|
type FieldRow struct {
|
|
Field string `json:"field"`
|
|
RowID uint64 `json:"rowID"`
|
|
RowKey string `json:"rowKey,omitempty"`
|
|
}
|
|
|
|
func (fr FieldRow) String() string {
|
|
return fmt.Sprintf("%s.%d", fr.Field, fr.RowID)
|
|
}
|
|
|
|
// TODO: we shouldn't need to string this
|
|
func uniqueGroupString(fr []FieldRow) string {
|
|
s := []string{}
|
|
for _, f := range fr {
|
|
s = append(s, f.String())
|
|
}
|
|
return strings.Join(s, "-")
|
|
}
|
|
|
|
// gbi is a groupBy item.
|
|
type gbi struct {
|
|
row *Row
|
|
fieldRow FieldRow
|
|
}
|
|
|
|
type GroupCount struct {
|
|
Group []FieldRow `json:"group"`
|
|
Count uint64 `json:"count"`
|
|
}
|
|
|
|
func mergeGroupCounts(gc, other []GroupCount) []GroupCount {
|
|
m := make(map[string]int)
|
|
for i := range gc {
|
|
m[uniqueGroupString(gc[i].Group)] = i
|
|
}
|
|
for i := range other {
|
|
if idx, found := m[uniqueGroupString(other[i].Group)]; found {
|
|
gc[idx].Count += other[i].Count
|
|
} else {
|
|
gc = append(gc, other[i])
|
|
}
|
|
}
|
|
return gc
|
|
}
|
|
|
|
func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, shard uint64) ([]GroupCount, error) {
|
|
// Fetch index.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
// fieldDirective is a combination of field
|
|
// instructions, represented as a string with
|
|
// the form [fieldName:offset:limit] or
|
|
// [fieldName:limit].
|
|
fieldDirectives, ok := c.Args["fields"]
|
|
if !ok {
|
|
return nil, errors.Wrap(ErrFieldsArgumentRequired, "executeGroupBy")
|
|
}
|
|
// Ensure that fieldDirectives is a list.
|
|
if _, ok := fieldDirectives.([]interface{}); !ok {
|
|
return nil, errors.Wrap(ErrExpectedFieldListArgument, "executeGroupBy")
|
|
}
|
|
// getFieldName extracts the fieldName portion of the
|
|
// fieldDirective.
|
|
getFieldName := func(s string) string {
|
|
parts := strings.Split(s, ":")
|
|
return parts[0]
|
|
}
|
|
// Ensure that all of the fields exist.
|
|
for _, fieldDirective := range fieldDirectives.([]interface{}) {
|
|
fieldName := getFieldName(fieldDirective.(string))
|
|
f := e.Holder.Field(index, fieldName)
|
|
if f == nil {
|
|
return nil, errors.Wrap(ErrFieldNotFound, fmt.Sprintf("executeGroupBy: %s", fieldDirective.(string)))
|
|
}
|
|
}
|
|
|
|
results := make([]GroupCount, 0)
|
|
var work [][]gbi
|
|
for _, fieldDirective := range fieldDirectives.([]interface{}) {
|
|
fieldName := getFieldName(fieldDirective.(string))
|
|
// Fetch fragment.
|
|
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
|
if frag == nil { // this means this whole shard doesn't have all it needs to continue
|
|
return results, nil
|
|
}
|
|
// Get filter based on the field directive.
|
|
filter, err := getGroupByFilterFunction(fieldDirective.(string))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
set := make([]gbi, 0)
|
|
for _, rowID := range frag.rowsWithFilter(0, filter) {
|
|
set = append(set, gbi{
|
|
row: frag.row(rowID),
|
|
fieldRow: FieldRow{
|
|
Field: fieldName,
|
|
RowID: rowID,
|
|
},
|
|
})
|
|
}
|
|
work = append(work, set)
|
|
}
|
|
for _, group := range product(work) {
|
|
group.gCnt.Count = group.row.Count()
|
|
if group.gCnt.Count > 0 {
|
|
results = append(results, group.gCnt)
|
|
}
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
// ppi is a product process item.
|
|
type ppi struct {
|
|
row *Row
|
|
gCnt GroupCount
|
|
}
|
|
|
|
// product generates the cartesian product of the input
|
|
// using tail recursion.
|
|
func product(input [][]gbi) []ppi {
|
|
if len(input) == 0 { // base return empty list
|
|
return []ppi{
|
|
{gCnt: GroupCount{Group: make([]FieldRow, 0)}},
|
|
}
|
|
}
|
|
|
|
res := make([]ppi, 0)
|
|
head := input[0] // take first element of the list
|
|
tail := product(input[1:]) // invoke product on remaining element
|
|
for h := range head { // for each head
|
|
for t := range tail { // iterate over the tail
|
|
s := ppi{gCnt: GroupCount{Group: make([]FieldRow, 0)}}
|
|
s.gCnt.Group = append([]FieldRow{head[h].fieldRow}, tail[t].gCnt.Group...) // had to insert at the front to match input order
|
|
if tail[t].row != nil { // first time around nothing to intersect
|
|
s.row = head[h].row.Intersect(tail[t].row)
|
|
} else {
|
|
s.row = head[h].row
|
|
}
|
|
res = append(res, s)
|
|
}
|
|
}
|
|
return res
|
|
}
|
|
|
|
func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) {
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
return e.executeRowsShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Determine limit so we can use it when reducing.
|
|
limit := int(^uint(0) >> 1)
|
|
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
|
|
return nil, err
|
|
} else if hasLimit {
|
|
limit = int(lim)
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.(RowIDs)
|
|
return other.merge(v.(RowIDs), limit)
|
|
}
|
|
// Get full result set.
|
|
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
results, _ := other.(RowIDs)
|
|
return results, nil
|
|
}
|
|
|
|
func (e *executor) executeRowsShard(ctx context.Context, index string, c *pql.Call, shard uint64) (RowIDs, error) {
|
|
// Fetch index.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
// Fetch field name from argument.
|
|
fieldName, ok := c.Args["field"].(string)
|
|
if !ok {
|
|
return nil, errors.New("Rows() argument required: field")
|
|
}
|
|
// Fetch field.
|
|
f := e.Holder.Field(index, fieldName)
|
|
if f == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
|
if frag == nil {
|
|
return make(RowIDs, 0), nil
|
|
}
|
|
|
|
start := uint64(0)
|
|
if previous, ok, err := c.UintArg("previous"); err != nil {
|
|
return nil, errors.Wrap(err, "getting previous")
|
|
} else if ok {
|
|
start = previous + 1
|
|
}
|
|
fmt.Println("calculated start is", start)
|
|
|
|
filter := noFilter
|
|
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
|
|
return nil, errors.Wrap(err, "getting limit")
|
|
} else if hasLimit {
|
|
filter = (&filterWithLimit{limit: limit}).filter
|
|
}
|
|
|
|
if columnID, ok, err := c.UintArg("column"); err != nil {
|
|
return nil, err
|
|
} else if ok {
|
|
return frag.rowsForColumnWithFilter(start, columnID, filter), nil
|
|
} else {
|
|
return frag.rowsWithFilter(start, filter), nil
|
|
}
|
|
}
|
|
|
|
// getGroupByFilterFunction returns a rowFilter based on the
|
|
// field directive provided.
|
|
func getGroupByFilterFunction(fieldDirective string) (rowFilter, error) {
|
|
parts := strings.Split(fieldDirective, ":")
|
|
hasLimit := false
|
|
hasOffset := false
|
|
limit := uint64(0)
|
|
offset := uint64(0)
|
|
var err error
|
|
// fieldDirective can have one of the following forms:
|
|
// [fieldName]
|
|
// [fieldName:limit]
|
|
// [fieldName:offset:limit]
|
|
//
|
|
// Note that a field directive with the form
|
|
// [fieldName:offset:limit:extra] will be treated as
|
|
// [fieldName:offset:limit] (i.e. `extra` is ignored).
|
|
if len(parts) == 1 {
|
|
return noFilter, nil
|
|
} else if len(parts) == 2 {
|
|
hasLimit = true
|
|
if limit, err = strconv.ParseUint(parts[1], 10, 64); err != nil {
|
|
return nil, errors.Wrap(err, "getting groupby field limit only value")
|
|
}
|
|
} else {
|
|
hasOffset = true
|
|
if offset, err = strconv.ParseUint(parts[1], 10, 64); err != nil {
|
|
return nil, errors.Wrap(err, "getting groupby field offset value")
|
|
}
|
|
if parts[2] != "" {
|
|
hasLimit = true
|
|
if limit, err = strconv.ParseUint(parts[2], 10, 64); err != nil {
|
|
return nil, errors.Wrap(err, "getting groupby field limit value")
|
|
}
|
|
}
|
|
}
|
|
if hasOffset && hasLimit {
|
|
f := filterWithOffsetLimit{offset: offset, limit: limit}
|
|
return f.filter, nil
|
|
} else if hasLimit {
|
|
f := filterWithLimit{limit: limit}
|
|
return f.filter, nil
|
|
}
|
|
f := filterWithOffset{offset: offset}
|
|
return f.filter, nil
|
|
}
|
|
|
|
func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
// Fetch index.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
|
|
// Fetch field name from argument.
|
|
fieldName, err := c.FieldArg()
|
|
if err != nil {
|
|
return nil, errors.New("Row() argument required: field")
|
|
}
|
|
f := e.Holder.Field(index, fieldName)
|
|
if f == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
rowID, rowOK, rowErr := c.UintArg(fieldName)
|
|
if rowErr != nil {
|
|
return nil, fmt.Errorf("Row() error with arg for row: %v", rowErr)
|
|
}
|
|
if !rowOK {
|
|
return nil, fmt.Errorf("Row() must specify %v", rowLabel)
|
|
}
|
|
|
|
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
|
if frag == nil {
|
|
return NewRow(), nil
|
|
}
|
|
return frag.row(rowID), nil
|
|
}
|
|
|
|
// executeIntersectShard executes a intersect() call for a local shard.
|
|
func (e *executor) executeIntersectShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
var other *Row
|
|
if len(c.Children) == 0 {
|
|
return nil, fmt.Errorf("empty Intersect query is currently not supported")
|
|
}
|
|
for i, input := range c.Children {
|
|
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if i == 0 {
|
|
other = row
|
|
} else {
|
|
other = other.Intersect(row)
|
|
}
|
|
}
|
|
other.invalidateCount()
|
|
return other, nil
|
|
}
|
|
|
|
// executeRangeShard executes a range() call for a local shard.
|
|
func (e *executor) executeRangeShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
// Handle bsiGroup ranges differently.
|
|
if c.HasConditionArg() {
|
|
return e.executeBSIGroupRangeShard(ctx, index, c, shard)
|
|
}
|
|
|
|
// Parse field.
|
|
fieldName, err := c.FieldArg()
|
|
if err != nil {
|
|
return nil, errors.New("Range() argument required: field")
|
|
}
|
|
|
|
// Retrieve column label.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return nil, ErrIndexNotFound
|
|
}
|
|
|
|
// Retrieve base field.
|
|
f := idx.Field(fieldName)
|
|
if f == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
// Read row & column id.
|
|
rowID, rowOK, err := c.UintArg(fieldName)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("executeRangeShard - reading row: %v", err)
|
|
}
|
|
if !rowOK {
|
|
return nil, fmt.Errorf("Range() must specify %q", rowLabel)
|
|
}
|
|
|
|
// Parse start time.
|
|
startTimeStr, ok := c.Args["_start"].(string)
|
|
if !ok {
|
|
return nil, errors.New("Range() start time required")
|
|
}
|
|
startTime, err := time.Parse(TimeFormat, startTimeStr)
|
|
if err != nil {
|
|
return nil, errors.New("cannot parse Range() start time")
|
|
}
|
|
|
|
// Parse end time.
|
|
endTimeStr, ok := c.Args["_end"].(string)
|
|
if !ok {
|
|
return nil, errors.New("Range() end time required")
|
|
}
|
|
endTime, err := time.Parse(TimeFormat, endTimeStr)
|
|
if err != nil {
|
|
return nil, errors.New("cannot parse Range() end time")
|
|
}
|
|
|
|
// If no quantum exists then return an empty bitmap.
|
|
q := f.TimeQuantum()
|
|
if q == "" {
|
|
return &Row{}, nil
|
|
}
|
|
|
|
// Union bitmaps across all time-based views.
|
|
row := &Row{}
|
|
for _, view := range viewsByTimeRange(viewStandard, startTime, endTime, q) {
|
|
f := e.Holder.fragment(index, fieldName, view, shard)
|
|
if f == nil {
|
|
continue
|
|
}
|
|
row = row.Union(f.row(rowID))
|
|
}
|
|
f.Stats.Count("range", 1, 1.0)
|
|
return row, nil
|
|
}
|
|
|
|
// executeBSIGroupRangeShard executes a range(bsiGroup) call for a local shard.
|
|
func (e *executor) executeBSIGroupRangeShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
// Only one conditional should be present.
|
|
if len(c.Args) == 0 {
|
|
return nil, errors.New("Range(): condition required")
|
|
} else if len(c.Args) > 1 {
|
|
return nil, errors.New("Range(): too many arguments")
|
|
}
|
|
|
|
// Extract conditional.
|
|
var fieldName string
|
|
var cond *pql.Condition
|
|
for k, v := range c.Args {
|
|
vv, ok := v.(*pql.Condition)
|
|
if !ok {
|
|
return nil, fmt.Errorf("Range(): %q: expected condition argument, got %v", k, v)
|
|
}
|
|
fieldName, cond = k, vv
|
|
}
|
|
|
|
f := e.Holder.Field(index, fieldName)
|
|
if f == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
// EQ null (not implemented: flip frag.NotNull with max ColumnID)
|
|
// NEQ null frag.NotNull()
|
|
// BETWEEN a,b(in) BETWEEN/frag.RangeBetween()
|
|
// BETWEEN a,b(out) BETWEEN/frag.NotNull()
|
|
// EQ <int> frag.RangeOp
|
|
// NEQ <int> frag.RangeOp
|
|
|
|
// Handle `!= null`.
|
|
if cond.Op == pql.NEQ && cond.Value == nil {
|
|
// Find bsiGroup.
|
|
bsig := f.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return nil, ErrBSIGroupNotFound
|
|
}
|
|
|
|
// Retrieve fragment.
|
|
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if frag == nil {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
return frag.notNull(bsig.BitDepth())
|
|
|
|
} else if cond.Op == pql.BETWEEN {
|
|
|
|
predicates, err := cond.IntSliceValue()
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "getting condition value")
|
|
}
|
|
|
|
// Only support two integers for the between operation.
|
|
if len(predicates) != 2 {
|
|
return nil, errors.New("Range(): BETWEEN condition requires exactly two integer values")
|
|
}
|
|
|
|
// The reason we don't just call:
|
|
// return f.RangeBetween(fieldName, predicates[0], predicates[1])
|
|
// here is because we need the call to be shard-specific.
|
|
|
|
// Find bsiGroup.
|
|
bsig := f.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return nil, ErrBSIGroupNotFound
|
|
}
|
|
|
|
baseValueMin, baseValueMax, outOfRange := bsig.baseValueBetween(predicates[0], predicates[1])
|
|
if outOfRange {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
// Retrieve fragment.
|
|
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if frag == nil {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
// If the query is asking for the entire valid range, just return
|
|
// the not-null bitmap for the bsiGroup.
|
|
if predicates[0] <= bsig.Min && predicates[1] >= bsig.Max {
|
|
return frag.notNull(bsig.BitDepth())
|
|
}
|
|
|
|
return frag.rangeBetween(bsig.BitDepth(), baseValueMin, baseValueMax)
|
|
|
|
} else {
|
|
|
|
// Only support integers for now.
|
|
value, ok := cond.Value.(int64)
|
|
if !ok {
|
|
return nil, errors.New("Range(): conditions only support integer values")
|
|
}
|
|
|
|
// Find bsiGroup.
|
|
bsig := f.bsiGroup(fieldName)
|
|
if bsig == nil {
|
|
return nil, ErrBSIGroupNotFound
|
|
}
|
|
|
|
baseValue, outOfRange := bsig.baseValue(cond.Op, value)
|
|
if outOfRange && cond.Op != pql.NEQ {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
// Retrieve fragment.
|
|
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
|
if frag == nil {
|
|
return NewRow(), nil
|
|
}
|
|
|
|
// LT[E] and GT[E] should return all not-null if selected range fully encompasses valid bsiGroup range.
|
|
if (cond.Op == pql.LT && value > bsig.Max) || (cond.Op == pql.LTE && value >= bsig.Max) ||
|
|
(cond.Op == pql.GT && value < bsig.Min) || (cond.Op == pql.GTE && value <= bsig.Min) {
|
|
return frag.notNull(bsig.BitDepth())
|
|
}
|
|
|
|
// outOfRange for NEQ should return all not-null.
|
|
if outOfRange && cond.Op == pql.NEQ {
|
|
return frag.notNull(bsig.BitDepth())
|
|
}
|
|
|
|
f.Stats.Count("range:bsigroup", 1, 1.0)
|
|
return frag.rangeOp(cond.Op, bsig.BitDepth(), baseValue)
|
|
}
|
|
}
|
|
|
|
// executeUnionShard executes a union() call for a local shard.
|
|
func (e *executor) executeUnionShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
other := NewRow()
|
|
for i, input := range c.Children {
|
|
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if i == 0 {
|
|
other = row
|
|
} else {
|
|
other = other.Union(row)
|
|
}
|
|
}
|
|
other.invalidateCount()
|
|
return other, nil
|
|
}
|
|
|
|
// executeXorShard executes a xor() call for a local shard.
|
|
func (e *executor) executeXorShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
|
other := NewRow()
|
|
for i, input := range c.Children {
|
|
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if i == 0 {
|
|
other = row
|
|
} else {
|
|
other = other.Xor(row)
|
|
}
|
|
}
|
|
other.invalidateCount()
|
|
return other, nil
|
|
}
|
|
|
|
// 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) {
|
|
if len(c.Children) == 0 {
|
|
return nil, errors.New("Not() requires an input row")
|
|
} else if len(c.Children) > 1 {
|
|
return nil, errors.New("Not() only accepts a single row input")
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return existenceRow.Difference(row), nil
|
|
}
|
|
|
|
// executeCount executes a count() call.
|
|
func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (uint64, error) {
|
|
if len(c.Children) == 0 {
|
|
return 0, errors.New("Count() requires an input bitmap")
|
|
} else if len(c.Children) > 1 {
|
|
return 0, errors.New("Count() only accepts a single bitmap input")
|
|
}
|
|
|
|
// Execute calls in bulk on each remote node and merge.
|
|
mapFn := func(shard uint64) (interface{}, error) {
|
|
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return row.Count(), nil
|
|
}
|
|
|
|
// Merge returned results at coordinating node.
|
|
reduceFn := func(prev, v interface{}) interface{} {
|
|
other, _ := prev.(uint64)
|
|
return other + v.(uint64)
|
|
}
|
|
|
|
result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
n, _ := result.(uint64)
|
|
|
|
return n, nil
|
|
}
|
|
|
|
// executeClearBit executes a Clear() call.
|
|
func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
|
|
fieldName, err := c.FieldArg()
|
|
if err != nil {
|
|
return false, errors.New("Clear() argument required: field")
|
|
}
|
|
|
|
// Retrieve field.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return false, ErrIndexNotFound
|
|
}
|
|
f := idx.Field(fieldName)
|
|
if f == nil {
|
|
return false, ErrFieldNotFound
|
|
}
|
|
|
|
// Read fields using labels.
|
|
rowID, ok, err := c.UintArg(fieldName)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading Clear() row: %v", err)
|
|
} else if !ok {
|
|
return false, fmt.Errorf("Clear() row argument '%v' required", rowLabel)
|
|
}
|
|
|
|
colID, ok, err := c.UintArg("_" + columnLabel)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading Clear() column: %v", err)
|
|
} else if !ok {
|
|
return false, fmt.Errorf("Clear() col argument '%v' required", columnLabel)
|
|
}
|
|
|
|
return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
|
|
}
|
|
|
|
// executeClearBitField executes a Clear() call for a single view.
|
|
func (e *executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *execOptions) (bool, error) {
|
|
shard := colID / ShardWidth
|
|
ret := false
|
|
for _, node := range e.Cluster.shardNodes(index, shard) {
|
|
// Update locally if host matches.
|
|
if node.ID == e.Node.ID {
|
|
val, err := f.ClearBit(rowID, colID)
|
|
if err != nil {
|
|
return false, err
|
|
} else if val {
|
|
ret = true
|
|
}
|
|
continue
|
|
}
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
continue
|
|
}
|
|
|
|
// Forward call to remote node otherwise.
|
|
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
|
|
return false, err
|
|
} else {
|
|
ret = res[0].(bool)
|
|
}
|
|
}
|
|
return ret, nil
|
|
}
|
|
|
|
// executeSet executes a Set() call.
|
|
func (e *executor) executeSet(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
|
|
|
|
// Read colID.
|
|
colID, ok, err := c.UintArg("_" + columnLabel)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading Set() column: %v", err)
|
|
} else if !ok {
|
|
return false, fmt.Errorf("Set() column argument '%v' required", columnLabel)
|
|
}
|
|
|
|
// Read field name.
|
|
fieldName, err := c.FieldArg()
|
|
if err != nil {
|
|
return false, errors.New("Set() argument required: field")
|
|
}
|
|
|
|
// Retrieve field.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return false, ErrIndexNotFound
|
|
}
|
|
f := idx.Field(fieldName)
|
|
if f == nil {
|
|
return false, ErrFieldNotFound
|
|
}
|
|
|
|
// Set column on existence field.
|
|
if ef := idx.existenceField(); ef != nil {
|
|
if _, err := ef.SetBit(0, colID, nil); err != nil {
|
|
return false, errors.Wrap(err, "setting existence column")
|
|
}
|
|
}
|
|
|
|
if f.Type() == FieldTypeInt {
|
|
// Read remaining fields using labels.
|
|
rowVal, ok, err := c.IntArg(fieldName)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading Set() row: %v", err)
|
|
} else if !ok {
|
|
return false, fmt.Errorf("Set() row argument '%v' required", rowLabel)
|
|
}
|
|
|
|
return e.executeSetValueField(ctx, index, c, f, colID, rowVal, opt)
|
|
} else {
|
|
// Read remaining fields using labels.
|
|
rowID, ok, err := c.UintArg(fieldName)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading Set() row: %v", err)
|
|
} else if !ok {
|
|
return false, fmt.Errorf("Set() row argument '%v' required", rowLabel)
|
|
}
|
|
|
|
var timestamp *time.Time
|
|
sTimestamp, ok := c.Args["_timestamp"].(string)
|
|
if ok {
|
|
t, err := time.Parse(TimeFormat, sTimestamp)
|
|
if err != nil {
|
|
return false, fmt.Errorf("invalid date: %s", sTimestamp)
|
|
}
|
|
timestamp = &t
|
|
}
|
|
|
|
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
|
|
}
|
|
}
|
|
|
|
// executeSetBitField executes a Set() call for a specific field.
|
|
func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *execOptions) (bool, error) {
|
|
shard := colID / ShardWidth
|
|
ret := false
|
|
|
|
for _, node := range e.Cluster.shardNodes(index, shard) {
|
|
// Update locally if host matches.
|
|
if node.ID == e.Node.ID {
|
|
val, err := f.SetBit(rowID, colID, timestamp)
|
|
if err != nil {
|
|
return false, err
|
|
} else if val {
|
|
ret = true
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
continue
|
|
}
|
|
|
|
// Forward call to remote node otherwise.
|
|
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
|
|
return false, err
|
|
} else {
|
|
ret = res[0].(bool)
|
|
}
|
|
}
|
|
return ret, nil
|
|
}
|
|
|
|
// executeSetValueField executes a Set() call for a specific int field.
|
|
func (e *executor) executeSetValueField(ctx context.Context, index string, c *pql.Call, f *Field, colID uint64, value int64, opt *execOptions) (bool, error) {
|
|
shard := colID / ShardWidth
|
|
ret := false
|
|
|
|
for _, node := range e.Cluster.shardNodes(index, shard) {
|
|
// Update locally if host matches.
|
|
if node.ID == e.Node.ID {
|
|
val, err := f.SetValue(colID, value)
|
|
if err != nil {
|
|
return false, err
|
|
} else if val {
|
|
ret = true
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
continue
|
|
}
|
|
|
|
// Forward call to remote node otherwise.
|
|
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
|
|
return false, err
|
|
} else {
|
|
ret = res[0].(bool)
|
|
}
|
|
}
|
|
return ret, nil
|
|
}
|
|
|
|
// executeSetRowAttrs executes a SetRowAttrs() call.
|
|
func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
|
|
fieldName, ok := c.Args["_field"].(string)
|
|
if !ok {
|
|
return errors.New("SetRowAttrs() field required")
|
|
}
|
|
|
|
// Retrieve field.
|
|
field := e.Holder.Field(index, fieldName)
|
|
if field == nil {
|
|
return ErrFieldNotFound
|
|
}
|
|
|
|
// Parse labels.
|
|
rowID, ok, err := c.UintArg("_" + rowLabel)
|
|
if err != nil {
|
|
return fmt.Errorf("reading SetRowAttrs() row: %v", err)
|
|
} else if !ok {
|
|
return fmt.Errorf("SetRowAttrs() row field '%v' required", rowLabel)
|
|
}
|
|
|
|
// Copy args and remove reserved fields.
|
|
attrs := pql.CopyArgs(c.Args)
|
|
delete(attrs, "_field")
|
|
delete(attrs, "_"+rowLabel)
|
|
|
|
// Set attributes.
|
|
if err := field.RowAttrStore().SetAttrs(rowID, attrs); err != nil {
|
|
return err
|
|
}
|
|
field.Stats.Count("SetRowAttrs", 1, 1.0)
|
|
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
return nil
|
|
}
|
|
|
|
// Execute on remote nodes in parallel.
|
|
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
|
|
resp := make(chan error, len(nodes))
|
|
for _, node := range nodes {
|
|
go func(node *Node) {
|
|
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil)
|
|
resp <- err
|
|
}(node)
|
|
}
|
|
|
|
// Return first error.
|
|
for range nodes {
|
|
if err := <-resp; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// executeBulkSetRowAttrs executes a set of SetRowAttrs() calls.
|
|
func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, calls []*pql.Call, opt *execOptions) ([]interface{}, error) {
|
|
// Collect attributes by field/id.
|
|
m := make(map[string]map[uint64]map[string]interface{})
|
|
for _, c := range calls {
|
|
field, ok := c.Args["_field"].(string)
|
|
if !ok {
|
|
return nil, errors.New("SetRowAttrs() field required")
|
|
}
|
|
|
|
// Retrieve field.
|
|
f := e.Holder.Field(index, field)
|
|
if f == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
rowID, ok, err := c.UintArg("_" + rowLabel)
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "reading SetRowAttrs() row")
|
|
} else if !ok {
|
|
return nil, fmt.Errorf("SetRowAttrs row field '%v' required", rowLabel)
|
|
}
|
|
|
|
// Copy args and remove reserved fields.
|
|
attrs := pql.CopyArgs(c.Args)
|
|
delete(attrs, "_field")
|
|
delete(attrs, "_"+rowLabel)
|
|
|
|
// Create field group, if not exists.
|
|
fieldMap := m[field]
|
|
if fieldMap == nil {
|
|
fieldMap = make(map[uint64]map[string]interface{})
|
|
m[field] = fieldMap
|
|
}
|
|
|
|
// Set or merge attributes.
|
|
attr := fieldMap[rowID]
|
|
if attr == nil {
|
|
fieldMap[rowID] = cloneAttrs(attrs)
|
|
} else {
|
|
for k, v := range attrs {
|
|
attr[k] = v
|
|
}
|
|
}
|
|
}
|
|
|
|
// Bulk insert attributes by field.
|
|
for name, fieldMap := range m {
|
|
// Retrieve field.
|
|
field := e.Holder.Field(index, name)
|
|
if field == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
// Set attributes.
|
|
if err := field.RowAttrStore().SetBulkAttrs(fieldMap); err != nil {
|
|
return nil, err
|
|
}
|
|
field.Stats.Count("SetRowAttrs", 1, 1.0)
|
|
}
|
|
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
return make([]interface{}, len(calls)), nil
|
|
}
|
|
|
|
// Execute on remote nodes in parallel.
|
|
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
|
|
resp := make(chan error, len(nodes))
|
|
for _, node := range nodes {
|
|
go func(node *Node) {
|
|
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: calls}, nil)
|
|
resp <- err
|
|
}(node)
|
|
}
|
|
|
|
// Return first error.
|
|
for range nodes {
|
|
if err := <-resp; err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// Return a set of nil responses to match the non-optimized return.
|
|
return make([]interface{}, len(calls)), nil
|
|
}
|
|
|
|
// executeSetColumnAttrs executes a SetColumnAttrs() call.
|
|
func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
|
|
// Retrieve index.
|
|
idx := e.Holder.Index(index)
|
|
if idx == nil {
|
|
return ErrIndexNotFound
|
|
}
|
|
|
|
col, okCol, errCol := c.UintArg("_" + columnLabel)
|
|
if errCol != nil || !okCol {
|
|
return fmt.Errorf("reading SetColumnAttrs() col errs: %v found %v", errCol, okCol)
|
|
}
|
|
|
|
// Copy args and remove reserved fields.
|
|
attrs := pql.CopyArgs(c.Args)
|
|
delete(attrs, "_"+columnLabel)
|
|
delete(attrs, "field")
|
|
|
|
// Set attributes.
|
|
if err := idx.ColumnAttrStore().SetAttrs(col, attrs); err != nil {
|
|
return err
|
|
}
|
|
idx.Stats.Count("SetProfileAttrs", 1, 1.0)
|
|
// Do not forward call if this is already being forwarded.
|
|
if opt.Remote {
|
|
return nil
|
|
}
|
|
|
|
// Execute on remote nodes in parallel.
|
|
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
|
|
resp := make(chan error, len(nodes))
|
|
for _, node := range nodes {
|
|
go func(node *Node) {
|
|
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil)
|
|
resp <- err
|
|
}(node)
|
|
}
|
|
|
|
// Return first error.
|
|
for range nodes {
|
|
if err := <-resp; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// remoteExec executes a PQL query remotely for a set of shards on a node.
|
|
func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *pql.Query, shards []uint64) (results []interface{}, err error) { // nolint: interfacer
|
|
// Encode request object.
|
|
pbreq := &QueryRequest{
|
|
Query: q.String(),
|
|
Shards: shards,
|
|
Remote: true,
|
|
}
|
|
|
|
pb, err := e.client.QueryNode(ctx, &node.URI, index, pbreq)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return pb.Results, pb.Err
|
|
}
|
|
|
|
// shardsByNode returns a mapping of nodes to shards.
|
|
// Returns errShardUnavailable if a shard cannot be allocated to a node.
|
|
func (e *executor) shardsByNode(nodes []*Node, index string, shards []uint64) (map[*Node][]uint64, error) {
|
|
m := make(map[*Node][]uint64)
|
|
|
|
loop:
|
|
for _, shard := range shards {
|
|
for _, node := range e.Cluster.shardNodes(index, shard) {
|
|
if Nodes(nodes).Contains(node) {
|
|
m[node] = append(m[node], shard)
|
|
continue loop
|
|
}
|
|
}
|
|
return nil, errShardUnavailable
|
|
}
|
|
return m, nil
|
|
}
|
|
|
|
// mapReduce maps and reduces data across the cluster.
|
|
//
|
|
// If a mapping of shards to a node fails then the shards are resplit across
|
|
// secondary nodes and retried. This continues to occur until all nodes are exhausted.
|
|
func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
|
|
ch := make(chan mapResponse)
|
|
|
|
// Wrap context with a cancel to kill goroutines on exit.
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
|
|
// If this is the coordinating node then start with all nodes in the cluster.
|
|
//
|
|
// However, if this request is being sent from the coordinator then all
|
|
// processing should be done locally so we start with just the local node.
|
|
var nodes []*Node
|
|
if !opt.Remote {
|
|
nodes = Nodes(e.Cluster.nodes).Clone()
|
|
} else {
|
|
nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)}
|
|
}
|
|
|
|
// Start mapping across all primary owners.
|
|
if err := e.mapper(ctx, ch, nodes, index, shards, c, opt, mapFn, reduceFn); err != nil {
|
|
return nil, errors.Wrap(err, "starting mapper")
|
|
}
|
|
|
|
// Iterate over all map responses and reduce.
|
|
var result interface{}
|
|
var shardN int
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, errors.Wrap(ctx.Err(), "context done")
|
|
case resp := <-ch:
|
|
// On error retry against remaining nodes. If an error returns then
|
|
// the context will cancel and cause all open goroutines to return.
|
|
|
|
if resp.err != nil {
|
|
// Filter out unavailable nodes.
|
|
nodes = Nodes(nodes).Filter(resp.node)
|
|
|
|
// Begin mapper against secondary nodes.
|
|
if err := e.mapper(ctx, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
|
|
return nil, resp.err
|
|
} else if err != nil {
|
|
return nil, errors.Wrap(err, "calling mapper")
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Reduce value.
|
|
result = reduceFn(result, resp.result)
|
|
|
|
// If all shards have been processed then return.
|
|
shardN += len(resp.shards)
|
|
if shardN >= len(shards) {
|
|
return result, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
|
|
// Group shards together by nodes.
|
|
m, err := e.shardsByNode(nodes, index, shards)
|
|
if err != nil {
|
|
return errors.Wrap(err, "shards by node")
|
|
}
|
|
|
|
// Execute each node in a separate goroutine.
|
|
for n, nodeShards := range m {
|
|
go func(n *Node, nodeShards []uint64) {
|
|
resp := mapResponse{node: n, shards: nodeShards}
|
|
|
|
// Send local shards to mapper, otherwise remote exec.
|
|
if n.ID == e.Node.ID {
|
|
resp.result, resp.err = e.mapperLocal(ctx, nodeShards, mapFn, reduceFn)
|
|
} else if !opt.Remote {
|
|
results, err := e.remoteExec(ctx, n, index, &pql.Query{Calls: []*pql.Call{c}}, nodeShards)
|
|
if len(results) > 0 {
|
|
resp.result = results[0]
|
|
}
|
|
resp.err = err
|
|
}
|
|
|
|
// Return response to the channel.
|
|
select {
|
|
case <-ctx.Done():
|
|
case ch <- resp:
|
|
}
|
|
}(n, nodeShards)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// mapperLocal performs map & reduce entirely on the local node.
|
|
func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
|
|
ch := make(chan mapResponse, len(shards))
|
|
|
|
for _, shard := range shards {
|
|
go func(shard uint64) {
|
|
result, err := mapFn(shard)
|
|
|
|
// Return response to the channel.
|
|
select {
|
|
case <-ctx.Done():
|
|
case ch <- mapResponse{result: result, err: err}:
|
|
}
|
|
}(shard)
|
|
}
|
|
|
|
// Reduce results
|
|
var maxShard int
|
|
var result interface{}
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case resp := <-ch:
|
|
if resp.err != nil {
|
|
return nil, resp.err
|
|
}
|
|
result = reduceFn(result, resp.result)
|
|
maxShard++
|
|
}
|
|
|
|
// Exit once all shards are processed.
|
|
if maxShard == len(shards) {
|
|
return result, nil
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
|
|
var colKey, rowKey, fieldName string
|
|
if c.Name == "Set" || c.Name == "Clear" || c.Name == "Row" {
|
|
// Positional args in new PQL syntax require special handling here.
|
|
colKey = "_" + columnLabel
|
|
fieldName, _ = c.FieldArg()
|
|
rowKey = fieldName
|
|
} else if c.Name == "SetRowAttrs" {
|
|
// Positional args in new PQL syntax require special handling here.
|
|
rowKey = "_" + rowLabel
|
|
fieldName = callArgString(c, "_field")
|
|
} else if c.Name == "Rows" {
|
|
fieldName = callArgString(c, "field")
|
|
rowKey = "previous"
|
|
} else {
|
|
colKey = "col"
|
|
fieldName = callArgString(c, "field")
|
|
rowKey = "row"
|
|
}
|
|
// Translate column key.
|
|
if idx.Keys() {
|
|
if c.Args[colKey] != nil && !isString(c.Args[colKey]) {
|
|
return errors.New("column value must be a string when index 'keys' option enabled")
|
|
}
|
|
if value := callArgString(c, colKey); value != "" {
|
|
ids, err := e.TranslateStore.TranslateColumnsToUint64(index, []string{value})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.Args[colKey] = ids[0]
|
|
}
|
|
} else {
|
|
if isString(c.Args[colKey]) {
|
|
return errors.New("string 'col' value not allowed unless index 'keys' option enabled")
|
|
}
|
|
}
|
|
|
|
// Translate row key, if field is specified & key exists.
|
|
if fieldName != "" {
|
|
field := idx.Field(fieldName)
|
|
if field == nil {
|
|
// Instead of returning ErrFieldNotFound here,
|
|
// we just return, and don't attempt the translation.
|
|
// The assumption is that the non-existent field
|
|
// will raise an error downstream when it's used.
|
|
return nil
|
|
}
|
|
if field.keys() {
|
|
if c.Args[rowKey] != nil && !isString(c.Args[rowKey]) {
|
|
return errors.New("row value must be a string when field 'keys' option enabled")
|
|
}
|
|
if value := callArgString(c, rowKey); value != "" {
|
|
ids, err := e.TranslateStore.TranslateRowsToUint64(index, fieldName, []string{value})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.Args[rowKey] = ids[0]
|
|
}
|
|
} else {
|
|
if isString(c.Args[rowKey]) {
|
|
return errors.New("string 'row' value not allowed unless field 'keys' option enabled")
|
|
}
|
|
}
|
|
}
|
|
|
|
// Translate child calls.
|
|
for _, child := range c.Children {
|
|
if err := e.translateCall(index, idx, child); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
|
|
switch result := result.(type) {
|
|
case *Row:
|
|
if idx.Keys() {
|
|
other := &Row{Attrs: result.Attrs}
|
|
for _, segment := range result.Segments() {
|
|
for _, col := range segment.Columns() {
|
|
key, err := e.TranslateStore.TranslateColumnToString(index, col)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
other.Keys = append(other.Keys, key)
|
|
}
|
|
}
|
|
return other, nil
|
|
}
|
|
|
|
case []Pair:
|
|
if fieldName := callArgString(call, "_field"); fieldName != "" {
|
|
field := idx.Field(fieldName)
|
|
if field == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
if field.keys() {
|
|
other := make([]Pair, len(result))
|
|
for i := range result {
|
|
key, err := e.TranslateStore.TranslateRowToString(index, fieldName, result[i].ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
other[i] = Pair{Key: key, Count: result[i].Count}
|
|
}
|
|
return other, nil
|
|
}
|
|
}
|
|
|
|
case []GroupCount:
|
|
other := make([]GroupCount, 0)
|
|
for _, gl := range result {
|
|
|
|
group := make([]FieldRow, len(gl.Group))
|
|
for i, g := range gl.Group {
|
|
group[i] = g
|
|
|
|
// TODO: It may be useful to cache this field lookup.
|
|
field := idx.Field(g.Field)
|
|
if field == nil {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
if field.keys() {
|
|
key, err := e.TranslateStore.TranslateRowToString(index, g.Field, g.RowID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
group[i].RowKey = key
|
|
}
|
|
}
|
|
|
|
other = append(other, GroupCount{
|
|
Group: group,
|
|
Count: gl.Count,
|
|
})
|
|
}
|
|
return other, nil
|
|
|
|
case RowIDs:
|
|
other := RowIdentifiers{}
|
|
|
|
fieldName := callArgString(call, "field")
|
|
if fieldName == "" {
|
|
return nil, ErrFieldNotFound
|
|
}
|
|
|
|
if field := idx.Field(fieldName); field == nil {
|
|
return nil, ErrFieldNotFound
|
|
} else if field.keys() {
|
|
other.Keys = make([]string, len(result))
|
|
for i, id := range result {
|
|
key, err := e.TranslateStore.TranslateRowToString(index, fieldName, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
other.Keys[i] = key
|
|
}
|
|
} else {
|
|
other.Rows = result
|
|
}
|
|
|
|
return other, nil
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// errShardUnavailable is a marker error if no nodes are available.
|
|
var errShardUnavailable = errors.New("shard unavailable")
|
|
|
|
type mapFunc func(shard uint64) (interface{}, error)
|
|
|
|
type reduceFunc func(prev, v interface{}) interface{}
|
|
|
|
type mapResponse struct {
|
|
node *Node
|
|
shards []uint64
|
|
|
|
result interface{}
|
|
err error
|
|
}
|
|
|
|
// execOptions represents an execution context for a single Execute() call.
|
|
type execOptions struct {
|
|
Remote bool
|
|
ExcludeRowAttrs bool
|
|
ExcludeColumns bool
|
|
ColumnAttrs bool
|
|
}
|
|
|
|
// hasOnlySetRowAttrs returns true if calls only contains SetRowAttrs() calls.
|
|
func hasOnlySetRowAttrs(calls []*pql.Call) bool {
|
|
if len(calls) == 0 {
|
|
return false
|
|
}
|
|
|
|
for _, call := range calls {
|
|
if call.Name != "SetRowAttrs" {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func needsShards(calls []*pql.Call) bool {
|
|
if len(calls) == 0 {
|
|
return false
|
|
}
|
|
for _, call := range calls {
|
|
switch call.Name {
|
|
case "Clear", "Set", "SetRowAttrs", "SetColumnAttrs":
|
|
continue
|
|
case "Count", "TopN", "Rows":
|
|
return true
|
|
// default catches Bitmap calls
|
|
default:
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// ValCount represents a grouping of sum & count for Sum() and Average() calls.
|
|
type ValCount struct {
|
|
Val int64 `json:"value"`
|
|
Count int64 `json:"count"`
|
|
}
|
|
|
|
func (vc *ValCount) add(other ValCount) ValCount {
|
|
return ValCount{
|
|
Val: vc.Val + other.Val,
|
|
Count: vc.Count + other.Count,
|
|
}
|
|
}
|
|
|
|
// smaller returns the smaller of the two ValCounts.
|
|
func (vc *ValCount) smaller(other ValCount) ValCount {
|
|
if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) {
|
|
return other
|
|
}
|
|
return ValCount{
|
|
Val: vc.Val,
|
|
Count: vc.Count,
|
|
}
|
|
}
|
|
|
|
// larger returns the larger of the two ValCounts.
|
|
func (vc *ValCount) larger(other ValCount) ValCount {
|
|
if vc.Count == 0 || (other.Val > vc.Val && other.Count > 0) {
|
|
return other
|
|
}
|
|
return ValCount{
|
|
Val: vc.Val,
|
|
Count: vc.Count,
|
|
}
|
|
}
|
|
|
|
func callArgString(call *pql.Call, key string) string {
|
|
value, ok := call.Args[key]
|
|
if !ok {
|
|
return ""
|
|
}
|
|
s, _ := value.(string)
|
|
return s
|
|
}
|
|
|
|
func isString(v interface{}) bool {
|
|
_, ok := v.(string)
|
|
return ok
|
|
}
|
|
|
|
// Filters to be used with RowsWithFilter queries.
|
|
type filterWithOffsetLimit struct {
|
|
offset, limit uint64
|
|
}
|
|
|
|
func (fol *filterWithOffsetLimit) filter(rowID uint64) (bool, bool) {
|
|
if rowID >= fol.offset {
|
|
if fol.limit > 0 {
|
|
fol.limit--
|
|
return true, false
|
|
}
|
|
return false, true
|
|
}
|
|
return false, false
|
|
}
|
|
|
|
type filterWithOffset struct {
|
|
offset uint64
|
|
}
|
|
|
|
func (fo *filterWithOffset) filter(rowID uint64) (bool, bool) {
|
|
return rowID >= fo.offset, false
|
|
}
|
|
|
|
type filterWithLimit struct {
|
|
limit uint64
|
|
}
|
|
|
|
func (fl *filterWithLimit) filter(rowID uint64) (bool, bool) { // nolint: unparam
|
|
if fl.limit > 0 {
|
|
fl.limit--
|
|
return true, false
|
|
}
|
|
return false, true
|
|
}
|