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

Allow All() to be called at the shard level

Browse source
This commit is contained in:
Travis 2019-12-20 22:43:50 -06:00
parent 7b30b91448
commit 586a13e942
2 changed files with 32 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
executor.go
View file

@ -720,7 +720,7 @@ func (e *executor) executeAllCall(ctx context.Context, index string, c *pql.Call
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)
return e.executeAllCallShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
@ -1113,6 +1113,8 @@ func (e *executor) executeBitmapCallShard(ctx context.Context, index string, c *
return e.executeNotShard(ctx, index, c, shard)
case "Shift":
return e.executeShiftShard(ctx, index, c, shard)
case "All": // Allow a shard computation to use All() (note, limit/offset not applied)
return e.executeAllCallShard(ctx, index, c, shard)
case "Precomputed":
return e.executePrecomputedCallShard(ctx, index, c, shard)
default:
@ -2524,9 +2526,9 @@ 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")
// executeAllCallShard executes an All() call for a local shard.
func (e *executor) executeAllCallShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeAllCallShard")
defer span.Finish()
if len(c.Children) > 0 {

26
executor_test.go
View file

@ -3039,6 +3039,32 @@ func TestExecutor_Execute_All(t *testing.T) {
}
}
})
// Ensure that a query which uses All() at the shard level can call it.
t.Run("AllShard", 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})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `
Set(3001, f=3)
Set(5001, f=5)
Set(5002, f=5)
`}); err != nil {
t.Fatalf("querying remote: %v", err)
}
expCols := []uint64{5001, 5002}
if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "Intersect(All(), Row(f=5))"}); err != nil {
t.Fatal(err)
} else if cols := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(cols, expCols) {
t.Fatalf("unexpected columns, got: %v, but expected: %v", cols, expCols)
}
})
}
// Ensure a row can be cleared.