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

thread contexts better through executor

Browse source 
When a mapper hits an error, we want it to immediately tell the
other things in that same mapper that they can stop now. But we
don't want to propagate that all the way back up; if a specific
node has a failure executing a query, we will in some cases want
to send a new query to other backup nodes, so the overall
context isn't cancelled yet.

In general, mapFn and reduceFn have been closures that inherit
a context from the function defining them -- but we don't want
that! We want them to be stopped if their specific mapper gets
cancelled, too, because otherwise they can consume a lot of
resources long after the mapper has stopped being interested
in them.  So now those are parameters passed into them,
and mapperLocal puts *those* contexts in the jobs shoved into
the job queue, and the workers pass the context in to the
mapFn/reduceFn.

We also check responses from reduceFn now; both mapReduce
and mapperLocal check for a possible error, and return that,
and reduce functions doing anything nontrivial check their
context.

We also add a few more explicit checks for context cancellation
in various places, especially in the GroupByIterator which is
what bit us that one time. The explicit check against ctx.Err
is officially safe as of Go 1.9 or so. (It was previously
unspecified, but on further study, the Go team concluded that
no actual implementation did anything else, and existing code
was already depending on that.) This also affects the rows
function, because that could potentially take quite a while to
run for a large fragment.
This commit is contained in:
Seebs 2020-05-28 14:22:33 -05:00
parent 33a90fd328
commit 439c710ca9
3 changed files with 131 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

168
executor.go
View file

@ -358,7 +358,9 @@ func (e *executor) handlePreCalls(ctx context.Context, index string, c *pql.Call
row = r.Pos
default:
return fmt.Errorf("precomputed call %s returned unexpected non-Row data: %T", c.Name, v)
}
if err := ctx.Err(); err != nil {
return err
}
c.Children = []*pql.Call{}
c.Name = "Precomputed"
@ -377,6 +379,9 @@ func (e *executor) handlePreCalls(ctx context.Context, index string, c *pql.Call
// handlePreCallChildren handles any pre-calls in the children of a given call.
func (e *executor) handlePreCallChildren(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) error {
for i := range c.Children {
if err := ctx.Err(); err != nil {
return err
}
if err := e.handlePreCalls(ctx, index, c.Children[i], shards, opt); err != nil {
return err
}
@ -384,6 +389,9 @@ func (e *executor) handlePreCallChildren(ctx context.Context, index string, c *p
for _, val := range c.Args {
// Handle Call() operations which exist inside named arguments, too.
if call, ok := val.(*pql.Call); ok {
if err := ctx.Err(); err != nil {
return err
}
if err := e.handlePreCalls(ctx, index, call, shards, opt); err != nil {
return err
}
@ -650,12 +658,12 @@ func (e *executor) executeIncludesColumnCall(ctx context.Context, index string,
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeIncludesColumnCallShard(ctx, index, c, shard, col)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(bool)
return other || v.(bool)
}
@ -705,12 +713,12 @@ func (e *executor) executeFieldValueCall(ctx context.Context, index string, c *p
shard := colID / ShardWidth
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeFieldValueCallShard(ctx, field, colID, shard)
}
// Select single returned result at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(ValCount)
if other.Count == 1 {
return other
@ -834,12 +842,15 @@ func (e *executor) executeAllCall(ctx context.Context, index string, c *pql.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) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeAllCallShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
if err := ctx.Err(); err != nil {
return err
}
other, _ := prev.(*Row)
if other == nil {
other = NewRow()
@ -889,12 +900,12 @@ func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, sh
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeSumCountShard(ctx, index, c, nil, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(ValCount)
return other.add(v.(ValCount))
}
@ -941,13 +952,16 @@ func (e *executor) executeGenericField(ctx context.Context, index string, c *pql
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeGenericFieldShard(ctx, index, c, op, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(SignedRow)
if err := ctx.Err(); err != nil {
return err
}
return other.union(v.(SignedRow))
}
@ -974,12 +988,12 @@ func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, sh
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeMinShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(ValCount)
return other.smaller(v.(ValCount))
}
@ -1010,12 +1024,12 @@ func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, sh
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeMaxShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(ValCount)
return other.larger(v.(ValCount))
}
@ -1042,12 +1056,12 @@ func (e *executor) executeMinRow(ctx context.Context, index string, c *pql.Call,
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeMinRowShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
// if minRowID exists, and if it is smaller than the other one return it.
// otherwise return the minRowID of the one which exists.
if prev == nil {
@ -1081,12 +1095,12 @@ func (e *executor) executeMaxRow(ctx context.Context, index string, c *pql.Call,
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeMaxRowShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
// if minRowID exists, and if it is smaller than the other one return it.
// otherwise return the minRowID of the one which exists.
if prev == nil {
@ -1138,16 +1152,19 @@ func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeBitmapCallShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(*Row)
if other == nil {
other = NewRow()
}
if err := ctx.Err(); err != nil {
return err
}
other.Merge(v.(*Row))
return other
}
@ -1505,12 +1522,12 @@ func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.C
defer span.Finish()
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeTopNShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(*PairsField)
vpf, _ := v.(*PairsField)
if other == nil {
@ -1518,6 +1535,9 @@ func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.C
} else if vpf == nil {
return other
}
if err := ctx.Err(); err != nil {
return err
}
other.Pairs = Pairs(other.Pairs).Add(vpf.Pairs)
return other
}
@ -1780,12 +1800,15 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeGroupByShard(ctx, index, c, filter, shard, childRows)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.([]GroupCount)
if err := ctx.Err(); err != nil {
return err
}
return mergeGroupCounts(other, v.([]GroupCount), limit)
}
// Get full result set.
@ -2199,7 +2222,7 @@ func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, s
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeRowsShard(ctx, index, fieldName, c, shard)
}
@ -2212,8 +2235,11 @@ func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, s
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(RowIDs)
if err := ctx.Err(); err != nil {
return err
}
return other.merge(v.(RowIDs), limit)
}
// Get full result set.
@ -2225,7 +2251,7 @@ func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, s
return results, nil
}
func (e *executor) executeRowsShard(_ context.Context, index string, fieldName string, c *pql.Call, shard uint64) (RowIDs, error) {
func (e *executor) executeRowsShard(ctx context.Context, index string, fieldName string, c *pql.Call, shard uint64) (RowIDs, error) {
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
@ -2335,12 +2361,15 @@ func (e *executor) executeRowsShard(_ context.Context, index string, fieldName s
}
for _, view := range views {
if err := ctx.Err(); err != nil {
return nil, err
}
frag := e.Holder.fragment(index, fieldName, view, shard)
if frag == nil {
continue
}
viewRows := frag.rows(start, filters...)
viewRows := frag.rows(ctx, start, filters...)
rowIDs = rowIDs.merge(viewRows, limit)
}
@ -2814,7 +2843,7 @@ func (e *executor) executeGenericCount(ctx context.Context, index string, c *pql
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
if err != nil {
return 0, err
@ -2823,7 +2852,7 @@ func (e *executor) executeGenericCount(ctx context.Context, index string, c *pql
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(uint64)
return other + v.(uint64)
}
@ -2849,7 +2878,7 @@ func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call,
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
if err != nil {
return 0, err
@ -2858,7 +2887,7 @@ func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call,
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
other, _ := prev.(uint64)
return other + v.(uint64)
}
@ -2971,12 +3000,12 @@ func (e *executor) executeClearRow(ctx context.Context, index string, c *pql.Cal
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeClearRowShard(ctx, index, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
val := v.(bool)
if prev == nil {
return val
@ -3059,12 +3088,12 @@ func (e *executor) executeSetRow(ctx context.Context, indexName string, c *pql.C
}
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeSetRowShard(ctx, indexName, c, shard)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
val := v.(bool)
if prev == nil {
return val
@ -3593,7 +3622,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
}
// Start mapping across all primary owners.
if err := e.mapper(ctx, ch, nodes, index, shards, c, opt, mapFn, reduceFn); err != nil {
if err := e.mapper(ctx, cancel, ch, nodes, index, shards, c, opt, mapFn, reduceFn); err != nil {
return nil, errors.Wrap(err, "starting mapper")
}
@ -3613,7 +3642,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
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 {
if err := e.mapper(ctx, cancel, 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")
@ -3622,7 +3651,11 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
}
// Reduce value.
result = reduceFn(result, resp.result)
result = reduceFn(ctx, result, resp.result)
if err, ok := result.(error); ok {
cancel()
return nil, err
}
// If all shards have been processed then return.
shardN += len(resp.shards)
@ -3670,9 +3703,10 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
return newRows
}
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 {
func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
defer span.Finish()
done := ctx.Done()
// Group shards together by nodes.
m, err := e.shardsByNode(nodes, index, shards)
@ -3699,11 +3733,16 @@ func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
}
resp.err = err
}
// Return response to the channel.
select {
case <-ctx.Done():
case <-done:
case ch <- resp:
// The cancel coming after the above send is intentional.
// We want to report the actual error that happened
// before we cause anything to return "context canceled".
if resp.err != nil {
cancel()
}
}
}(n, nodeShards)
}
@ -3720,7 +3759,7 @@ type job struct {
func worker(work chan job) {
for j := range work {
result, err := j.mapFn(j.shard)
result, err := j.mapFn(j.ctx, j.shard)
select {
case <-j.ctx.Done():
@ -3733,6 +3772,9 @@ func worker(work chan job) {
func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
defer span.Finish()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
done := ctx.Done()
ch := make(chan mapResponse, len(shards))
@ -3750,13 +3792,17 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
var result interface{}
for {
select {
case <-ctx.Done():
case <-done:
return nil, ctx.Err()
case resp := <-ch:
if resp.err != nil {
return nil, resp.err
}
result = reduceFn(result, resp.result)
result = reduceFn(ctx, result, resp.result)
if err, ok := result.(error); ok {
cancel()
return nil, err
}
maxShard++
}
@ -4268,9 +4314,9 @@ func validateQueryContext(ctx context.Context) error {
// errShardUnavailable is a marker error if no nodes are available.
var errShardUnavailable = errors.New("shard unavailable")
type mapFunc func(shard uint64) (interface{}, error)
type mapFunc func(ctx context.Context, shard uint64) (interface{}, error)
type reduceFunc func(prev, v interface{}) interface{}
type reduceFunc func(ctx context.Context, prev, v interface{}) interface{}
type mapResponse struct {
node *Node
@ -4770,18 +4816,21 @@ func newGroupByIterator(executor *executor, rowIDs []RowIDs, children []*pql.Cal
// nextAtIdx is a recursive helper method for getting the next row for the field
// at index i, and then updating the rows in the "higher" fields if it wraps.
func (gbi *groupByIterator) nextAtIdx(i int) {
func (gbi *groupByIterator) nextAtIdx(ctx context.Context, i int) (err error) {
// loop until we find a non-empty row. This is an optimization - the loop and if/break can be removed.
for {
if err = ctx.Err(); err != nil {
return err
}
nr, rowID, value, wrapped := gbi.rowIters[i].Next()
if nr == nil {
gbi.done = true
return
return nil
}
if wrapped && i != 0 {
gbi.nextAtIdx(i - 1)
if gbi.done {
return
err = gbi.nextAtIdx(ctx, i-1)
if gbi.done || err != nil {
return err
}
}
if i == 0 && gbi.filter != nil {
@ -4798,6 +4847,7 @@ func (gbi *groupByIterator) nextAtIdx(i int) {
break
}
}
return nil
}
// Next returns a GroupCount representing the next group by record. When there
@ -4805,6 +4855,9 @@ func (gbi *groupByIterator) nextAtIdx(i int) {
func (gbi *groupByIterator) Next(ctx context.Context) (ret GroupCount, done bool, err error) {
// loop until we find a result with count > 0
for {
if err := ctx.Err(); err != nil {
return ret, false, err
}
if gbi.done {
return ret, true, nil
}
@ -4832,7 +4885,10 @@ func (gbi *groupByIterator) Next(ctx context.Context) (ret GroupCount, done bool
}
}
if ret.Count == 0 {
gbi.nextAtIdx(len(gbi.rows) - 1)
err := gbi.nextAtIdx(ctx, len(gbi.rows)-1)
if err != nil {
return ret, false, err
}
continue
}
break
@ -4847,9 +4903,9 @@ func (gbi *groupByIterator) Next(ctx context.Context) (ret GroupCount, done bool
}
// set up for next call
gbi.nextAtIdx(len(gbi.rows) - 1)
err = gbi.nextAtIdx(ctx, len(gbi.rows)-1)
return ret, false, nil
return ret, false, err
}
// getCondIntSlice looks at the field, the cond op type (which is

16
fragment.go
View file

@ -2607,14 +2607,14 @@ func filterWithRows(rows []uint64) rowFilter {
// returning done == true will cause processing to stop after all filters for
// this container have been processed. The rows accumulated up to this point
// (including this row if all filters passed) will be returned.
func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
func (f *fragment) rows(ctx context.Context, start uint64, filters ...rowFilter) []uint64 {
f.mu.RLock()
defer f.mu.RUnlock()
return f.unprotectedRows(start, filters...)
return f.unprotectedRows(ctx, start, filters...)
}
// unprotectedRows calls rows without grabbing the mutex.
func (f *fragment) unprotectedRows(start uint64, filters ...rowFilter) []uint64 {
func (f *fragment) unprotectedRows(ctx context.Context, start uint64, filters ...rowFilter) []uint64 {
startKey := rowToKey(start)
i, _ := f.storage.Containers.Iterator(startKey)
rows := make([]uint64, 0)
@ -2622,6 +2622,10 @@ func (f *fragment) unprotectedRows(start uint64, filters ...rowFilter) []uint64
// Loop over the existing containers.
for i.Next() {
// caller doesn't need a result anymore.
if ctx.Err() != nil {
return nil
}
key, c := i.Value()
// virtual row for the current container
@ -2861,7 +2865,7 @@ type setRowIterator struct {
func (f *fragment) setRowIterator(wrap bool, filters ...rowFilter) rowIterator {
return &setRowIterator{
f: f,
rowIDs: f.rows(0, filters...), // TODO: this may be memory intensive in high cardinality cases
rowIDs: f.rows(context.Background(), 0, filters...), // TODO: this may be memory intensive in high cardinality cases
wrap: wrap,
}
}
@ -3282,7 +3286,7 @@ func newRowsVector(f *fragment) *rowsVector {
// otherwise it returns false. Ensure that you already
// have the mutex before calling this.
func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.unprotectedRows(0, filterColumn(colID))
rows := v.f.unprotectedRows(context.Background(), 0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {
@ -3316,7 +3320,7 @@ func newBoolVector(f *fragment) *boolVector {
// otherwise it returns false. Ensure that you already
// have the fragment mutex before calling this.
func (v *boolVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.unprotectedRows(0, filterColumn(colID))
rows := v.f.unprotectedRows(context.Background(), 0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {

17
fragment_internal_test.go
View file

@ -2412,8 +2412,9 @@ func TestGetZipfRowsSliceRoaring(t *testing.T) {
if err != nil {
t.Fatalf("importing roaring: %v", err)
}
if !reflect.DeepEqual(f.rows(0), []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) {
t.Fatalf("unexpected rows: %v", f.rows(0))
rows := f.rows(context.Background(), 0)
if !reflect.DeepEqual(rows, []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) {
t.Fatalf("unexpected rows: %v", rows)
}
for i := uint64(1); i < 10; i++ {
if f.row(i).Count() >= f.row(i-1).Count() {
@ -2714,12 +2715,12 @@ func TestFragment_RowsIteration(t *testing.T) {
}
}
ids := f.rows(0)
ids := f.rows(context.Background(), 0)
if !reflect.DeepEqual(expectedAll, ids) {
t.Fatalf("Do not match %v %v", expectedAll, ids)
}
ids = f.rows(0, filterColumn(1))
ids = f.rows(context.Background(), 0, filterColumn(1))
if !reflect.DeepEqual(expectedOdd, ids) {
t.Fatalf("Do not match %v %v", expectedOdd, ids)
}
@ -2738,12 +2739,12 @@ func TestFragment_RowsIteration(t *testing.T) {
t.Fatal(err)
}
ids := f.rows(0)
ids := f.rows(context.Background(), 0)
if !reflect.DeepEqual(expected, ids) {
t.Fatalf("Do not match %v %v", expected, ids)
}
ids = f.rows(0, filterColumn(66000))
ids = f.rows(context.Background(), 0, filterColumn(66000))
if !reflect.DeepEqual(expected, ids) {
t.Fatalf("Do not match %v %v", expected, ids)
}
@ -2761,11 +2762,11 @@ func TestFragment_RowsIteration(t *testing.T) {
t.Fatal(err)
}
ids := f.rows(0)
ids := f.rows(context.Background(), 0)
if !reflect.DeepEqual(expectedRows, ids) {
t.Fatalf("Do not match %v %v", expectedRows, ids)
}
ids = f.rows(0, filterColumn(c))
ids = f.rows(context.Background(), 0, filterColumn(c))
if !reflect.DeepEqual(expectedRows, ids) {
t.Fatalf("Do not match %v %v", expectedRows, ids)
}