make Distinct work across nodes, probably

Problem: A top-level bare "Distinct" call returns results only
for shards on the current node.

Analysis: We don't actually want to limit Distinct calls to "available"
shards at all. We just want to run them on everything. But we already
did that in generating the precomputed results; all we need to do is,
if we get a non-shard-specific request for precomputed values, just
return all the values.

It's pretty hard to create logic for this using our fancy mapReduce,
but also we could just... not do that.
This commit is contained in:
Seebs 2020-03-11 11:03:24 -05:00
parent 12ba11a437
commit a495b6c227

View file

@ -1006,37 +1006,14 @@ func (e *executor) executeMaxRow(ctx context.Context, index string, c *pql.Call,
// executePrecomputedCall pretends to execute a call that we have a precomputed value for.
func (e *executor) executePrecomputedCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executePrecomputedCall")
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executePrecomputedCall")
defer span.Finish()
result := NewRow()
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
if c.Precomputed != nil {
return c.Precomputed[shard], nil
}
// This might not be an error -- if there were no values, we will not have created
// the corresponding row.
return NewRow(), nil
for _, row := range c.Precomputed {
result.Merge(row.(*Row))
}
// 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")
}
row, _ := other.(*Row)
return row, nil
return result, nil
}
// executeBitmapCall executes a call that returns a bitmap.