From 78ff75690d1afe6eca272c76dacb2fe6f6a13c23 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 27 Sep 2018 16:14:40 -0500 Subject: [PATCH] convert Rows to use previous/limit 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. --- executor.go | 78 +++++++++++++++++++-------------------- executor_test.go | 4 +- fragment.go | 28 +++++++++----- fragment_internal_test.go | 6 +-- 4 files changed, 61 insertions(+), 55 deletions(-) diff --git a/executor.go b/executor.go index 5de820a22..d8d5fba2b 100644 --- a/executor.go +++ b/executor.go @@ -738,10 +738,10 @@ type RowIdentifiers struct { // the proto package needs access to it. type RowIDs []uint64 -func (r RowIDs) merge(other RowIDs) RowIDs { +func (r RowIDs) merge(other RowIDs, limit int) RowIDs { i, j := 0, 0 result := make(RowIDs, 0) - for i < len(r) && j < len(other) { + for i < len(r) && j < len(other) && len(result) < limit { av, bv := r[i], other[j] if av < bv { result = append(result, av) @@ -755,11 +755,11 @@ func (r RowIDs) merge(other RowIDs) RowIDs { j++ } } - for i < len(r) { + for i < len(r) && len(result) < limit { result = append(result, r[i]) i++ } - for j < len(other) { + for j < len(other) && len(result) < limit { result = append(result, other[j]) j++ } @@ -897,7 +897,7 @@ func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql } set := make([]gbi, 0) - for _, rowID := range frag.rowsWithFilter(filter) { + for _, rowID := range frag.rowsWithFilter(0, filter) { set = append(set, gbi{ row: frag.row(rowID), fieldRow: FieldRow{ @@ -955,10 +955,19 @@ func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, s 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)) + return other.merge(v.(RowIDs), limit) } // Get full result set. other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) @@ -966,22 +975,6 @@ func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, s return nil, err } results, _ := other.(RowIDs) - // 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 } @@ -1005,14 +998,29 @@ func (e *executor) executeRowsShard(ctx context.Context, index string, c *pql.Ca 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 { - // TODO: it's possible that filters could be applied here, so this returns too early. - return frag.rowsForColumn(columnID), nil + return frag.rowsForColumnWithFilter(start, columnID, filter), nil + } else { + return frag.rowsWithFilter(start, filter), nil } - filter := getFilterFunction(c) - return frag.rowsWithFilter(filter), nil } // getGroupByFilterFunction returns a rowFilter based on the @@ -1061,21 +1069,6 @@ func getGroupByFilterFunction(fieldDirective string) (rowFilter, error) { f := filterWithOffset{offset: offset} return f.filter, nil } -func getFilterFunction(c *pql.Call) rowFilter { - offset, hasOffset, _ := c.UintArg("shardoffset") - limit, hasLimit, _ := c.UintArg("shardlimit") - if hasOffset && hasLimit { - f := filterWithOffsetLimit{offset: offset, limit: limit} - return f.filter - } else if hasOffset { - f := filterWithOffset{offset: offset} - return f.filter - } else if hasLimit { - f := filterWithLimit{limit: limit} - return f.filter - } - return noFilter -} func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) { // Fetch index. @@ -2005,6 +1998,9 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error { // 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") diff --git a/executor_test.go b/executor_test.go index 73af2e2f8..0de3016f1 100644 --- a/executor_test.go +++ b/executor_test.go @@ -1639,7 +1639,7 @@ func benchmarkExistence(nn bool, b *testing.B) { func BenchmarkExecutor_Existence_True(b *testing.B) { benchmarkExistence(true, b) } func BenchmarkExecutor_Existence_False(b *testing.B) { benchmarkExistence(false, b) } -func TestExecutor_Execute_RowIDs(t *testing.T) { +func TestExecutor_Execute_Rows(t *testing.T) { c := test.MustRunCluster(t, 1) defer c.Close() hldr := test.Holder{Holder: c[0].Server.Holder()} @@ -1662,7 +1662,7 @@ func TestExecutor_Execute_RowIDs(t *testing.T) { t.Fatalf("unexpected columns: %+v", columns) } - if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general, offset=1,limit=2)`}); err != nil { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general, previous=10,limit=2)`}); err != nil { t.Fatal(err) } else if columns := res.Results[0].(pilosa.RowIdentifiers); !reflect.DeepEqual(columns, pilosa.RowIdentifiers{Rows: []uint64{11, 12}}) { t.Fatalf("unexpected columns: %+v", columns) diff --git a/fragment.go b/fragment.go index bcd8e9b13..6b0a53fe3 100644 --- a/fragment.go +++ b/fragment.go @@ -58,6 +58,9 @@ const ( // exponent. shardVsContainerExponent = shardWidthExponent - 16 + // width of roaring containers is 2^16 + containerWidth = 1 << 16 + // snapshotExt is the file extension used for an in-process snapshot. snapshotExt = ".snapshotting" @@ -1772,12 +1775,13 @@ var noFilter = func(rowID uint64) (bool, bool) { return true, false } // rows returns all rows by calling rowsWithFilter() // with a completely unrestrictive filter. -func (f *fragment) rows() []uint64 { - return f.rowsWithFilter(noFilter) +func (f *fragment) rows(start uint64) []uint64 { + return f.rowsWithFilter(start, noFilter) } -func (f *fragment) rowsWithFilter(filter rowFilter) []uint64 { - i, _ := f.storage.Containers.Iterator(0) +func (f *fragment) rowsWithFilter(start uint64, filter rowFilter) []uint64 { + startKey := rowToKey(start) + i, _ := f.storage.Containers.Iterator(startKey) rows := make([]uint64, 0) var lastRow uint64 = math.MaxUint64 @@ -1806,14 +1810,13 @@ func (f *fragment) rowsWithFilter(filter rowFilter) []uint64 { } -// rowsForColumn is similar to the rows method, but isolated -// to a single column. func (f *fragment) rowsForColumn(columnID uint64) []uint64 { - return f.rowsForColumnWithFilter(columnID, noFilter) + return f.rowsForColumnWithFilter(0, columnID, noFilter) } -func (f *fragment) rowsForColumnWithFilter(columnID uint64, filter rowFilter) []uint64 { - i, _ := f.storage.Containers.Iterator(0) +func (f *fragment) rowsForColumnWithFilter(start, columnID uint64, filter rowFilter) []uint64 { + startKey := rowToKey(start) + i, _ := f.storage.Containers.Iterator(startKey) rows := make([]uint64, 0) colID := columnID % ShardWidth @@ -2136,3 +2139,10 @@ func (v *rowsVector) Get(colID uint64) (uint64, bool) { // Set is not used for rowsVector. func (v *rowsVector) Set(colID, rowID uint64) {} + +// rowToKey converts a Pilosa row ID to the key of the container which starts +// that row in the bitmap which represents this entire fragment. A fragment is +// all the rows within a shard within a field concatenated together. +func rowToKey(rowID uint64) (key uint64) { + return rowID * (ShardWidth / containerWidth) +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index dd2d00b41..70b6ad78e 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1341,7 +1341,7 @@ func TestFragment_RowsIteration(t *testing.T) { } } - ids := f.rows() + ids := f.rows(0) if !reflect.DeepEqual(expectedAll, ids) { t.Fatalf("Do not match %v %v", expectedAll, ids) } @@ -1365,7 +1365,7 @@ func TestFragment_RowsIteration(t *testing.T) { t.Fatal(err) } - ids := f.rows() + ids := f.rows(0) if !reflect.DeepEqual(expected, ids) { t.Fatalf("Do not match %v %v", expected, ids) } @@ -1388,7 +1388,7 @@ func TestFragment_RowsIteration(t *testing.T) { t.Fatal(err) } - ids := f.rows() + ids := f.rows(0) if !reflect.DeepEqual(expectedRows, ids) { t.Fatalf("Do not match %v %v", expectedRows, ids) }