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

Merge pull request #408 from seebs/execontext

Browse source 
thread contexts better through executor
This commit is contained in:
seebs 2020-06-04 15:18:12 -05:00 • committed by GitHub
commit 33728e65d9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
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)
}