From ff153459ee41552234e5c2a010494aadd4eed5ba Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 17 Jul 2019 23:09:38 -0500 Subject: [PATCH 1/3] union many things at once to cut down allocations --- executor.go | 8 ++++--- row.go | 63 +++++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 52 insertions(+), 19 deletions(-) diff --git a/executor.go b/executor.go index 2110a92c6..57a206d90 100644 --- a/executor.go +++ b/executor.go @@ -1512,14 +1512,16 @@ func (e *executor) executeRowShard(ctx context.Context, index string, c *pql.Cal } // Union bitmaps across all time-based views. - row := &Row{} - for _, view := range viewsByTimeRange(viewStandard, fromTime, toTime, q) { + views := viewsByTimeRange(viewStandard, fromTime, toTime, q) + rows := make([]*Row, 0, len(views)) + for _, view := range views { f := e.Holder.fragment(index, fieldName, view, shard) if f == nil { continue } - row = row.Union(f.row(rowID)) + rows = append(rows, f.row(rowID)) } + row := rows[0].Union(rows[1:]...) f.Stats.Count("range", 1, 1.0) return row, nil diff --git a/row.go b/row.go index 79c918311..09a3390dc 100644 --- a/row.go +++ b/row.go @@ -150,21 +150,48 @@ func (r *Row) Xor(other *Row) *Row { } // Union returns the bitwise union of r and other. -func (r *Row) Union(other *Row) *Row { - var segments []rowSegment - itr := newMergeSegmentIterator(r.segments, other.segments) - for s0, s1 := itr.next(); s0 != nil || s1 != nil; s0, s1 = itr.next() { - if s1 == nil { - segments = append(segments, *s0) - continue - } else if s0 == nil { - segments = append(segments, *s1) - continue - } - segments = append(segments, *s0.Union(s1)) +func (r *Row) Union(others ...*Row) *Row { + segments := make([][]rowSegment, 0, len(others)+1) + if len(r.segments) > 0 { + segments = append(segments, r.segments) } - - return &Row{segments: segments} + nextSegs := make([][]rowSegment, 0, len(others)+1) + toProcess := make([]*rowSegment, 0, len(others)+1) + var output []rowSegment + for _, other := range others { + if len(other.segments) > 0 { + segments = append(segments, other.segments) + } + } + for len(segments) > 0 { + shard := segments[0][0].shard + for _, segs := range segments { + if segs[0].shard < shard { + shard = segs[0].shard + } + } + nextSegs = nextSegs[:0] + toProcess := toProcess[:0] + for _, segs := range segments { + if segs[0].shard == shard { + toProcess = append(toProcess, &segs[0]) + segs = segs[1:] + } + if len(segs) > 0 { + nextSegs = append(nextSegs, segs) + } + } + // at this point, "toProcess" is a list of all the segments + // sharing the lowest ID, and nextSegs is a list of all the others. + // Swap the segment lists (so we don't have to reallocate it) + segments, nextSegs = nextSegs, segments + if len(toProcess) == 1 { + output = append(output, *toProcess[0]) + } else { + output = append(output, *toProcess[0].Union(toProcess[1:]...)) + } + } + return &Row{segments: output} } // Difference returns the diff of r and other. @@ -350,8 +377,12 @@ func (s *rowSegment) Intersect(other *rowSegment) *rowSegment { } // Union returns the bitwise union of s and other. -func (s *rowSegment) Union(other *rowSegment) *rowSegment { - data := s.data.Union(other.data) +func (s *rowSegment) Union(others ...*rowSegment) *rowSegment { + datas := make([]*roaring.Bitmap, len(others)) + for i, other := range others { + datas[i] = other.data + } + data := s.data.Union(datas...) data.Freeze() return &rowSegment{ From 2c311e685eeff9df1f96a811c5c878f04024fd5b Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 18 Jul 2019 00:19:19 -0500 Subject: [PATCH 2/3] add bounds check when getting time rows --- executor.go | 5 +++++ executor_test.go | 2 ++ 2 files changed, 7 insertions(+) diff --git a/executor.go b/executor.go index 57a206d90..4d1d3e4c9 100644 --- a/executor.go +++ b/executor.go @@ -1521,6 +1521,11 @@ func (e *executor) executeRowShard(ctx context.Context, index string, c *pql.Cal } rows = append(rows, f.row(rowID)) } + if len(rows) == 0 { + return &Row{}, nil + } else if len(rows) == 1 { + return rows[0], nil + } row := rows[0].Union(rows[1:]...) f.Stats.Count("range", 1, 1.0) return row, nil diff --git a/executor_test.go b/executor_test.go index be10b80bb..75a07130e 100644 --- a/executor_test.go +++ b/executor_test.go @@ -3292,6 +3292,7 @@ func TestExecutor_Execute_RowsTime(t *testing.T) { `Rows(f)`, `Rows(f, from=2002-01-01T00:00)`, `Rows(f, to=2003-02-03T00:00)`, + `Rows(f, from=2002-01-01T00:00, to=2002-01-02T00:00)`, } expResults := [][]uint64{ {1}, @@ -3300,6 +3301,7 @@ func TestExecutor_Execute_RowsTime(t *testing.T) { {1, 2, 3, 4, 13}, {2, 3, 4, 13}, {1, 2, 3, 13}, + {2}, } responses := runCallTest(t, writeQuery, readQueries, From 5b655418b1878b0a3eaa91f5c16e14a94f95f547 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 18 Jul 2019 09:27:52 -0500 Subject: [PATCH 3/3] add another single day query test --- executor_test.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/executor_test.go b/executor_test.go index 75a07130e..76a8300bd 100644 --- a/executor_test.go +++ b/executor_test.go @@ -2920,6 +2920,7 @@ func TestExecutor_Execute_ClearRow(t *testing.T) { Set(2, f=10, 2001-01-01T00:00)` readQueries := []string{ `Row(f=1, from=1999-12-31T00:00, to=2003-01-01T03:00)`, + `Row(f=1, from=2002-01-01T00:00, to=2002-01-02T00:00)`, `ClearRow(f=1)`, `Row(f=1, from=1999-12-31T00:00, to=2003-01-01T03:00)`, `Row(f=10, from=1999-12-31T00:00, to=2003-01-01T03:00)`, @@ -2931,20 +2932,26 @@ func TestExecutor_Execute_ClearRow(t *testing.T) { t.Fatalf("unexpected columns: %+v", columns) } + // Single day query (regression test) + if columns := responses[1].Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{7}) { + t.Fatalf("unexpected columns: %+v", columns) + } + // Clear the row and ensure we get a `true` response. - if res := responses[1].Results[0].(bool); !res { + if res := responses[2].Results[0].(bool); !res { t.Fatalf("unexpected clear row result: %+v", res) } // Ensure the row is empty. - if columns := responses[2].Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{}) { + if columns := responses[3].Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{}) { t.Fatalf("unexpected columns: %+v", columns) } // Ensure other rows were not affected. - if columns := responses[3].Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{2}) { + if columns := responses[4].Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{2}) { t.Fatalf("unexpected columns: %+v", columns) } + }) t.Run("Int", func(t *testing.T) {