From f15cb9e05ca4a5b1098f2be67851d74d01811d0c Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 30 May 2019 15:15:08 +0300 Subject: [PATCH 01/11] added roaring min --- roaring/containers.go | 7 ++++ roaring/containers_btree.go | 7 ++++ roaring/containers_test.go | 44 ++++++++++++++++++++++++- roaring/roaring.go | 65 ++++++++++++++++++++++++++++++++++--- roaring/roaring_test.go | 26 +++++++++++++++ 5 files changed, 144 insertions(+), 5 deletions(-) diff --git a/roaring/containers.go b/roaring/containers.go index dc5d4118d..0cd515b9f 100644 --- a/roaring/containers.go +++ b/roaring/containers.go @@ -114,6 +114,13 @@ func (sc *sliceContainers) Clone() Containers { return other } +func (sc *sliceContainers) First() (key uint64, c *Container) { + if len(sc.keys) == 0 { + return 0, nil + } + return sc.keys[0], sc.containers[0] +} + func (sc *sliceContainers) Last() (key uint64, c *Container) { if len(sc.keys) == 0 { return 0, nil diff --git a/roaring/containers_btree.go b/roaring/containers_btree.go index 5934a1244..11f282a83 100644 --- a/roaring/containers_btree.go +++ b/roaring/containers_btree.go @@ -142,6 +142,13 @@ func (btc *bTreeContainers) Clone() Containers { return nbtc } +func (btc *bTreeContainers) First() (key uint64, c *Container) { + if btc.tree.Len() == 0 { + return 0, nil + } + return btc.tree.First() +} + func (btc *bTreeContainers) Last() (key uint64, c *Container) { if btc.tree.Len() == 0 { return 0, nil diff --git a/roaring/containers_test.go b/roaring/containers_test.go index ad95f2f79..6d159b73d 100644 --- a/roaring/containers_test.go +++ b/roaring/containers_test.go @@ -15,6 +15,7 @@ package roaring import ( + "reflect" "testing" ) @@ -98,5 +99,46 @@ func testContainersIterator(cs Containers, t *testing.T) { if itr.Next() { t.Fatalf("itr should be done, but got true") } - +} + +func TestContainers(t *testing.T) { + cs := NewFileBitmap().Containers + first := NewContainerArray([]uint16{1, 2, 3}) + last := NewContainerArray([]uint16{1, 2, 3, 4, 5, 6}) + cs.Put(3, first) + cs.Put(6, last) + + key, container := cs.First() + if key != 3 { + t.Fatalf("cs.First key 3 != %d", key) + } + if !reflect.DeepEqual(first, container) { + t.Fatalf("cs.First container %v != %v", first, container) + } + + key, container = cs.Last() + if key != 6 { + t.Fatalf("cs.First key 6 != %d", key) + } + if !reflect.DeepEqual(last, container) { + t.Fatalf("cs.First container %v != %v", last, container) + } + + cs = NewFileBitmap().Containers + + key, container = cs.First() + if key != 0 { + t.Fatalf("cs.First key 0 != %d", key) + } + if nil != container { + t.Fatalf("cs.First container nil != %v", container) + } + + key, container = cs.Last() + if key != 0 { + t.Fatalf("cs.First key 0 != %d", key) + } + if nil != container { + t.Fatalf("cs.First container nil != %v", container) + } } diff --git a/roaring/roaring.go b/roaring/roaring.go index 2d1a2d08a..73b7e3dc6 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -95,6 +95,9 @@ type Containers interface { // Clone does a deep copy of Containers, including cloning all containers contained. Clone() Containers + // First returns the lowest key and associated container. + First() (key uint64, c *Container) + // Last returns the highest key and associated container. Last() (key uint64, c *Container) @@ -320,6 +323,18 @@ func (b *Bitmap) remove(v uint64) bool { return c.remove(lowbits(v)) } +// Min returns the lowest value in the bitmap. +// Second return value is true if containers exist in the bitmap. +func (b *Bitmap) Min() (uint64, bool) { + if b.Containers.Size() == 0 { + return 0, false + } + + hb, c := b.Containers.First() + lb, ok := c.min() + return hb<<16 | uint64(lb), ok +} + // Max returns the highest value in the bitmap. // Returns zero if the bitmap is empty. func (b *Bitmap) Max() uint64 { @@ -1860,6 +1875,17 @@ func (c *Container) runRemove(v uint16) bool { return true } +// min returns the minimum value in the container. +func (c *Container) min() (uint16, bool) { + if c.isArray() { + return c.arrayMin() + } else if c.isRun() { + return c.runMin() + } else { + return c.bitmapMin() + } +} + // max returns the maximum value in the container. func (c *Container) max() uint16 { if c.isArray() { @@ -1871,6 +1897,15 @@ func (c *Container) max() uint16 { } } +// Second result value is true if array is non-empty. +func (c *Container) arrayMin() (uint16, bool) { + array := c.array() + if len(array) == 0 { + return 0, false + } + return array[0], true +} + func (c *Container) arrayMax() uint16 { array := c.array() if len(array) == 0 { @@ -1879,21 +1914,43 @@ func (c *Container) arrayMax() uint16 { return array[len(array)-1] } +// Second result value is true if array is non-empty. +func (c *Container) bitmapMin() (uint16, bool) { + bitmap := c.bitmap() + for i := 0; i < len(bitmap); i++ { + // If value is zero then skip. + v := bitmap[i] + if v != 0 { + r := bits.TrailingZeros64(v) + return uint16(r + i*64), true + } + } + return 0, false +} + func (c *Container) bitmapMax() uint16 { // Search bitmap in reverse order. bitmap := c.bitmap() - for i := len(bitmap); i > 0; i-- { + for i := len(bitmap) - 1; i > 0; i-- { // If value is zero then skip. - v := bitmap[i-1] + v := bitmap[i] if v != 0 { r := bits.LeadingZeros64(v) - return uint16((i-1)*64 + 63 - r) + return uint16(i*64 + 63 - r) } - } return 0 } +// Second result value is true if array is non-empty. +func (c *Container) runMin() (uint16, bool) { + runs := c.runs() + if len(runs) == 0 { + return 0, false + } + return runs[0].start, true +} + func (c *Container) runMax() uint16 { runs := c.runs() if len(runs) == 0 { diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index b4f792629..210ffa94d 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -309,6 +309,32 @@ func TestBitmap_Max(t *testing.T) { } } +// Ensure bitmap can return the lowest value. +func TestBitmap_Min(t *testing.T) { + bm := roaring.NewFileBitmap() + for i := uint64(100000); i > 0; i-- { + if _, err := bm.Add(i); err != nil { + t.Fatalf("adding bits: %v", err) + } + + v, ok := bm.Min() + if !ok { + t.Fatalf("ok should be true") + } + + if v != i { + t.Fatalf("min: got=%d; want=%d", v, i) + } + } + + // empty bitmap + bm = roaring.NewFileBitmap() + _, ok := bm.Min() + if ok { + t.Fatalf("ok should be false") + } +} + // Ensure CountRange is correct even if rangekey is prior to initial container. func TestBitmap_BitmapCountRangeEdgeCase(t *testing.T) { s := uint64(2009 * pilosa.ShardWidth) From 778ae1e8e25f8c1fb6b80dd555efed156a2cfd4b Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 30 May 2019 15:20:04 +0300 Subject: [PATCH 02/11] added enterprise btree first --- enterprise/b/containers_btree.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/enterprise/b/containers_btree.go b/enterprise/b/containers_btree.go index a051d7cde..eb7358a5d 100644 --- a/enterprise/b/containers_btree.go +++ b/enterprise/b/containers_btree.go @@ -149,6 +149,13 @@ func (btc *bTreeContainers) Clone() roaring.Containers { return nbtc } +func (btc *bTreeContainers) First() (key uint64, c *roaring.Container) { + if btc.tree.Len() == 0 { + return 0, nil + } + return btc.tree.First() +} + func (btc *bTreeContainers) Last() (key uint64, c *roaring.Container) { if btc.tree.Len() == 0 { return 0, nil From d2aca3bbfc643d60b2d6accd02a3b575cf4aa1c4 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 31 May 2019 15:32:15 +0300 Subject: [PATCH 03/11] Added MinRow and MaxRow calls --- executor.go | 141 ++++++++++++++++++++++++++++++++++++++++++++++++++++ fragment.go | 20 +++++++- go.sum | 1 + 3 files changed, 160 insertions(+), 2 deletions(-) diff --git a/executor.go b/executor.go index 607840a42..56785ad48 100644 --- a/executor.go +++ b/executor.go @@ -264,6 +264,12 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s case "Max": e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) return e.executeMax(ctx, index, c, shards, opt) + case "MinRow": + e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) + return e.executeMinRow(ctx, index, c, shards, opt) + case "MaxRow": + e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) + return e.executeMaxRow(ctx, index, c, shards, opt) case "Clear": return e.executeClearBit(ctx, index, c, opt) case "ClearRow": @@ -468,6 +474,74 @@ func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, sh return other, nil } +// executeMinRow executes a MinRow() call. +func (e *executor) executeMinRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRow") + defer span.Finish() + + if field := c.Args["field"]; field == "" { + return ValCount{}, errors.New("MinRow(): field required") + } + + // Execute calls in bulk on each remote node and merge. + mapFn := func(shard uint64) (interface{}, error) { + return e.executeMinRowShard(ctx, index, c, shard) + } + + // Merge returned results at coordinating node. + reduceFn := func(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. + prevp, _ := prev.(Pair) + vp, _ := v.(Pair) + if prevp.Count > 0 && vp.Count > 0 { + if prevp.ID < vp.ID { + return prevp + } + return vp + } else if prevp.Count > 0 { + return prevp + } + return vp + } + + return e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) +} + +// executeMinRow executes a MaxRow() call. +func (e *executor) executeMaxRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRow") + defer span.Finish() + + if field := c.Args["field"]; field == "" { + return ValCount{}, errors.New("MaxRow(): field required") + } + + // Execute calls in bulk on each remote node and merge. + mapFn := func(shard uint64) (interface{}, error) { + return e.executeMaxRowShard(ctx, index, c, shard) + } + + // Merge returned results at coordinating node. + reduceFn := func(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. + prevp, _ := prev.(Pair) + vp, _ := v.(Pair) + if prevp.Count > 0 && vp.Count > 0 { + if prevp.ID > vp.ID { + return prevp + } + return vp + } else if prevp.Count > 0 { + return prevp + } + return vp + } + + return e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) +} + // executeBitmapCall executes a call that returns a bitmap. func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCall") @@ -648,6 +722,32 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal }, nil } +// executeMinRowShard returns the minimum row ID for a shard. +func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRowShard") + defer span.Finish() + + fieldName, _ := c.Args["field"].(string) + field := e.Holder.Field(index, fieldName) + if field == nil { + return Pair{}, nil + } + + fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) + if fragment == nil { + return Pair{}, nil + } + + count := uint64(1) + if !fragment.hasRowID { + count = 0 + } + return Pair{ + ID: fragment.minRowID, + Count: count, + }, nil +} + // executeMaxShard calculates the max for bsiGroups on a shard. func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) { span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxShard") @@ -689,6 +789,32 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal }, nil } +// executeMaxRowShard returns the minimum row ID for a shard. +func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRowShard") + defer span.Finish() + + fieldName, _ := c.Args["field"].(string) + field := e.Holder.Field(index, fieldName) + if field == nil { + return Pair{}, nil + } + + fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) + if fragment == nil { + return Pair{}, nil + } + + count := uint64(1) + if !fragment.hasRowID { + count = 0 + } + return Pair{ + ID: fragment.maxRowID, + Count: count, + }, nil +} + // executeTopN executes a TopN() call. // This first performs the TopN() to determine the top results and then // requeries to retrieve the full counts for each of the top results. @@ -2624,6 +2750,21 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res return other, nil } + case Pair: + if fieldName := callArgString(call, "field"); fieldName != "" { + field := idx.Field(fieldName) + if field == nil { + return nil, fmt.Errorf("field %q not found", fieldName) + } + if field.keys() { + key, err := e.TranslateStore.TranslateRowToString(index, fieldName, result.ID) + if err != nil { + return nil, err + } + return Pair{Key: key, Count: result.Count}, nil + } + } + case []Pair: if fieldName := callArgString(call, "_field"); fieldName != "" { field := idx.Field(fieldName) diff --git a/fragment.go b/fragment.go index d9b9e8128..b162b965e 100644 --- a/fragment.go +++ b/fragment.go @@ -118,6 +118,8 @@ type fragment struct { // Stats reporting. maxRowID uint64 + minRowID uint64 + hasRowID bool // Cache containing full rows (not just counts). rowCache bitmapCache @@ -188,8 +190,10 @@ func (f *fragment) Open() error { f.checksums = make(map[int][]byte) // Read last bit to determine max row. - pos := f.storage.Max() - f.maxRowID = pos / ShardWidth + f.maxRowID = f.storage.Max() / ShardWidth + min, ok := f.storage.Min() + f.minRowID = min / ShardWidth + f.hasRowID = ok f.stats.Gauge("rows", float64(f.maxRowID), 1.0) return nil }(); err != nil { @@ -519,6 +523,10 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err f.maxRowID = rowID f.stats.Gauge("rows", float64(f.maxRowID), 1.0) } + if !f.hasRowID || rowID < f.minRowID { + f.minRowID = rowID + f.hasRowID = true + } return changed, nil } @@ -1023,6 +1031,14 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin return max, count } +func (f *fragment) rowIDMin() (uint64, bool) { + return f.minRowID, f.hasRowID +} + +func (f *fragment) rowIDMax() uint64 { + return f.maxRowID +} + // rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate. func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) { switch op { diff --git a/go.sum b/go.sum index 674e2adea..a66c2be51 100644 --- a/go.sum +++ b/go.sum @@ -86,6 +86,7 @@ github.com/shirou/gopsutil v2.18.12+incompatible h1:1eaJvGomDnH74/5cF4CTmTbLHAri github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U= github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc= +github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI= github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ= From dd728f28ede2a4372903a27d5b2d0c5acf4f2d3f Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 31 May 2019 17:13:25 +0300 Subject: [PATCH 04/11] remove unused code --- executor.go | 6 ------ fragment.go | 8 -------- 2 files changed, 14 deletions(-) diff --git a/executor.go b/executor.go index 56785ad48..12789c3dd 100644 --- a/executor.go +++ b/executor.go @@ -724,9 +724,6 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal // executeMinRowShard returns the minimum row ID for a shard. func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { - span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRowShard") - defer span.Finish() - fieldName, _ := c.Args["field"].(string) field := e.Holder.Field(index, fieldName) if field == nil { @@ -750,9 +747,6 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql. // executeMaxShard calculates the max for bsiGroups on a shard. func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) { - span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxShard") - defer span.Finish() - var filter *Row if len(c.Children) == 1 { row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard) diff --git a/fragment.go b/fragment.go index b162b965e..87e7915b1 100644 --- a/fragment.go +++ b/fragment.go @@ -1031,14 +1031,6 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin return max, count } -func (f *fragment) rowIDMin() (uint64, bool) { - return f.minRowID, f.hasRowID -} - -func (f *fragment) rowIDMax() uint64 { - return f.maxRowID -} - // rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate. func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) { switch op { From 592a191aee6b46dc353f5d9c49615cbcb76ee9f2 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 31 May 2019 17:21:44 +0300 Subject: [PATCH 05/11] translate key into Pair only for MinRow, MaxRow --- executor.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/executor.go b/executor.go index 12789c3dd..9f0ee8dac 100644 --- a/executor.go +++ b/executor.go @@ -2755,6 +2755,10 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res if err != nil { return nil, err } + if call.Name == "MinRow" || call.Name == "MaxRow" { + result.Key = key + return result, nil + } return Pair{Key: key, Count: result.Count}, nil } } From 1379cbbcd6f6fc56cd682c152a564c2d766075fa Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 31 May 2019 17:35:50 +0300 Subject: [PATCH 06/11] remove unused code --- executor.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/executor.go b/executor.go index 9f0ee8dac..8b07f9c9c 100644 --- a/executor.go +++ b/executor.go @@ -785,9 +785,6 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal // executeMaxRowShard returns the minimum row ID for a shard. func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { - span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRowShard") - defer span.Finish() - fieldName, _ := c.Args["field"].(string) field := e.Holder.Field(index, fieldName) if field == nil { From 8be7bd69569d6b0e4a7a3d7086084f559d2e7574 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Mon, 3 Jun 2019 13:56:50 +0300 Subject: [PATCH 07/11] add tests for MinRow and MaxRow --- executor.go | 46 +++++++++++------------ executor_test.go | 97 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 23 deletions(-) diff --git a/executor.go b/executor.go index 8b07f9c9c..3529995f5 100644 --- a/executor.go +++ b/executor.go @@ -722,29 +722,6 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal }, nil } -// executeMinRowShard returns the minimum row ID for a shard. -func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { - fieldName, _ := c.Args["field"].(string) - field := e.Holder.Field(index, fieldName) - if field == nil { - return Pair{}, nil - } - - fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) - if fragment == nil { - return Pair{}, nil - } - - count := uint64(1) - if !fragment.hasRowID { - count = 0 - } - return Pair{ - ID: fragment.minRowID, - Count: count, - }, nil -} - // executeMaxShard calculates the max for bsiGroups on a shard. func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) { var filter *Row @@ -783,6 +760,29 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal }, nil } +// executeMinRowShard returns the minimum row ID for a shard. +func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { + fieldName, _ := c.Args["field"].(string) + field := e.Holder.Field(index, fieldName) + if field == nil { + return Pair{}, nil + } + + fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) + if fragment == nil { + return Pair{}, nil + } + + count := uint64(1) + if !fragment.hasRowID { + count = 0 + } + return Pair{ + ID: fragment.minRowID, + Count: count, + }, nil +} + // executeMaxRowShard returns the minimum row ID for a shard. func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { fieldName, _ := c.Args["field"].(string) diff --git a/executor_test.go b/executor_test.go index 3565c40fd..be10b80bb 100644 --- a/executor_test.go +++ b/executor_test.go @@ -1415,6 +1415,103 @@ func TestExecutor_Execute_MinMax(t *testing.T) { }) } +// Ensure MinRow() and MaxRow() queries can be executed. +func TestExecutor_Execute_MinMaxRow(t *testing.T) { + t.Run("RowID", func(t *testing.T) { + c := test.MustRunCluster(t, 1) + defer c.Close() + hldr := test.Holder{Holder: c[0].Server.Holder()} + + idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}) + if err != nil { + t.Fatal(err) + } + + if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil { + t.Fatal(err) + } + + if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: ` + Set(0, f=7000) + Set(3, f=50) + Set(` + strconv.Itoa(ShardWidth+1) + `, f=10000) + Set(1000, f=1) + Set(` + strconv.Itoa(ShardWidth+2) + `, f=5000) + `}); err != nil { + t.Fatal(err) + } + + t.Run("MinRow", func(t *testing.T) { + result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MinRow(field=f)"}) + if err != nil { + t.Fatal(err) + } + target := pilosa.Pair{ID: 1, Count: 1} + if !reflect.DeepEqual(target, result.Results[0]) { + t.Fatalf("unexpected result %v != %v", target, result.Results[0]) + } + }) + + t.Run("MaxRow", func(t *testing.T) { + result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MaxRow(field=f)"}) + if err != nil { + t.Fatal(err) + } + target := pilosa.Pair{ID: 10000, Count: 1} + if !reflect.DeepEqual(target, result.Results[0]) { + t.Fatalf("unexpected result %v != %v", target, result.Results[0]) + } + }) + }) + + t.Run("RowKey", func(t *testing.T) { + c := test.MustRunCluster(t, 1) + defer c.Close() + hldr := test.Holder{Holder: c[0].Server.Holder()} + + idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}) + if err != nil { + t.Fatal(err) + } + + if _, err := idx.CreateField("f", pilosa.OptFieldKeys()); err != nil { + t.Fatal(err) + } + + if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: ` + Set(0, f="seven-thousand") + Set(3, f="fifty") + Set(` + strconv.Itoa(ShardWidth+1) + `, f="ten-thousand") + Set(1000, f="one") + Set(` + strconv.Itoa(ShardWidth+2) + `, f="five-thousand") + `}); err != nil { + t.Fatal(err) + } + + t.Run("MinRow", func(t *testing.T) { + result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MinRow(field=f)"}) + if err != nil { + t.Fatal(err) + } + target := pilosa.Pair{Key: "seven-thousand", ID: 1, Count: 1} + if !reflect.DeepEqual(target, result.Results[0]) { + t.Fatalf("unexpected result %v != %v", target, result.Results[0]) + } + }) + + t.Run("MaxRow", func(t *testing.T) { + result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MaxRow(field=f)"}) + if err != nil { + t.Fatal(err) + } + target := pilosa.Pair{Key: "five-thousand", ID: 5, Count: 1} + if !reflect.DeepEqual(target, result.Results[0]) { + t.Fatalf("unexpected result %v != %v", target, result.Results[0]) + } + }) + }) +} + // Ensure a Sum() query can be executed. func TestExecutor_Execute_Sum(t *testing.T) { t.Run("ColumnID", func(t *testing.T) { From b64a3e0c68f666067e5c3a67475f296bdc2758dc Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Mon, 3 Jun 2019 16:29:04 +0300 Subject: [PATCH 08/11] adds filter support to MinRow and MaxRow --- executor.go | 32 ++++++++++++++++++++++---------- fragment.go | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 10 deletions(-) diff --git a/executor.go b/executor.go index 3529995f5..7d474b6a4 100644 --- a/executor.go +++ b/executor.go @@ -762,6 +762,15 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal // executeMinRowShard returns the minimum row ID for a shard. func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { + var filter *Row + if len(c.Children) == 1 { + row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard) + if err != nil { + return Pair{}, err + } + filter = row + } + fieldName, _ := c.Args["field"].(string) field := e.Holder.Field(index, fieldName) if field == nil { @@ -773,18 +782,24 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql. return Pair{}, nil } - count := uint64(1) - if !fragment.hasRowID { - count = 0 - } + minRowID, count := fragment.minRow(filter) return Pair{ - ID: fragment.minRowID, + ID: minRowID, Count: count, }, nil } // executeMaxRowShard returns the minimum row ID for a shard. func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { + var filter *Row + if len(c.Children) == 1 { + row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard) + if err != nil { + return Pair{}, err + } + filter = row + } + fieldName, _ := c.Args["field"].(string) field := e.Holder.Field(index, fieldName) if field == nil { @@ -796,12 +811,9 @@ func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql. return Pair{}, nil } - count := uint64(1) - if !fragment.hasRowID { - count = 0 - } + maxRowID, count := fragment.maxRow(filter) return Pair{ - ID: fragment.maxRowID, + ID: maxRowID, Count: count, }, nil } diff --git a/fragment.go b/fragment.go index 642f378bb..c6cc03291 100644 --- a/fragment.go +++ b/fragment.go @@ -1036,6 +1036,46 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin return max, count } +// minRow returns minRowID of the rows in the filter and its count. +// if filter is nil, it returns fragment.minRowID, 1 +// if fragment has no rows, it returns 0, 0 +func (f *fragment) minRow(filter *Row) (uint64, uint64) { + if f.hasRowID { + if filter == nil { + return f.minRowID, 1 + } + // iterate from min row ID and return the first that intersects with filter. + for i := f.minRowID; i <= f.maxRowID; i++ { + row := f.row(i).Intersect(filter) + count := row.Count() + if count > 0 { + return i, count + } + } + } + return 0, 0 +} + +// maxRow returns maxRowID of the rows in the filter and its count. +// if filter is nil, it returns fragment.maxRowID, 1 +// if fragment has no rows, it returns 0, 0 +func (f *fragment) maxRow(filter *Row) (uint64, uint64) { + if f.hasRowID { + if filter == nil { + return f.maxRowID, 1 + } + // iterate back from max row ID and return the first that intersects with filter. + for i := f.maxRowID; i >= f.minRowID; i-- { + row := f.row(i).Intersect(filter) + count := row.Count() + if count > 0 { + return i, count + } + } + } + return 0, 0 +} + // rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate. func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) { switch op { From 13a42d9c0783fdf25e79b832611124d938c8cb23 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Tue, 11 Jun 2019 16:58:32 +0300 Subject: [PATCH 09/11] replaced min code with bmp.iterator --- executor.go | 2 +- roaring/containers_btree.go | 7 ----- roaring/containers_slice.go | 7 ----- roaring/containers_test.go | 43 ----------------------------- roaring/roaring.go | 55 ++----------------------------------- 5 files changed, 3 insertions(+), 111 deletions(-) diff --git a/executor.go b/executor.go index 7d474b6a4..7eb13dd79 100644 --- a/executor.go +++ b/executor.go @@ -789,7 +789,7 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql. }, nil } -// executeMaxRowShard returns the minimum row ID for a shard. +// executeMaxRowShard returns the maximum row ID for a shard. func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) { var filter *Row if len(c.Children) == 1 { diff --git a/roaring/containers_btree.go b/roaring/containers_btree.go index 2b1dd55a7..c2dbc01f5 100644 --- a/roaring/containers_btree.go +++ b/roaring/containers_btree.go @@ -164,13 +164,6 @@ func (btc *bTreeContainers) Freeze() Containers { return nbtc } -func (btc *bTreeContainers) First() (key uint64, c *Container) { - if btc.tree.Len() == 0 { - return 0, nil - } - return btc.tree.First() -} - func (btc *bTreeContainers) Last() (key uint64, c *Container) { if btc.tree.Len() == 0 { return 0, nil diff --git a/roaring/containers_slice.go b/roaring/containers_slice.go index 7a5722cf4..cbff4f179 100644 --- a/roaring/containers_slice.go +++ b/roaring/containers_slice.go @@ -131,13 +131,6 @@ func (sc *sliceContainers) Freeze() Containers { return other } -func (sc *sliceContainers) First() (key uint64, c *Container) { - if len(sc.keys) == 0 { - return 0, nil - } - return sc.keys[0], sc.containers[0] -} - func (sc *sliceContainers) Last() (key uint64, c *Container) { if len(sc.keys) == 0 { return 0, nil diff --git a/roaring/containers_test.go b/roaring/containers_test.go index 9d5e1610d..1869d31af 100644 --- a/roaring/containers_test.go +++ b/roaring/containers_test.go @@ -15,7 +15,6 @@ package roaring import ( - "reflect" "testing" ) @@ -100,45 +99,3 @@ func testContainersIterator(cs Containers, t *testing.T) { t.Fatalf("itr should be done, but got true") } } - -func TestContainers(t *testing.T) { - cs := NewFileBitmap().Containers - first := NewContainerArray([]uint16{1, 2, 3}) - last := NewContainerArray([]uint16{1, 2, 3, 4, 5, 6}) - cs.Put(3, first) - cs.Put(6, last) - - key, container := cs.First() - if key != 3 { - t.Fatalf("cs.First key 3 != %d", key) - } - if !reflect.DeepEqual(first, container) { - t.Fatalf("cs.First container %v != %v", first, container) - } - - key, container = cs.Last() - if key != 6 { - t.Fatalf("cs.First key 6 != %d", key) - } - if !reflect.DeepEqual(last, container) { - t.Fatalf("cs.First container %v != %v", last, container) - } - - cs = NewFileBitmap().Containers - - key, container = cs.First() - if key != 0 { - t.Fatalf("cs.First key 0 != %d", key) - } - if nil != container { - t.Fatalf("cs.First container nil != %v", container) - } - - key, container = cs.Last() - if key != 0 { - t.Fatalf("cs.First key 0 != %d", key) - } - if nil != container { - t.Fatalf("cs.First container nil != %v", container) - } -} diff --git a/roaring/roaring.go b/roaring/roaring.go index 89dca0f17..7b7b5d144 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -103,9 +103,6 @@ type Containers interface { // are shared (but marked as frozen). Freeze() Containers - // First returns the lowest key and associated container. - First() (key uint64, c *Container) - // Last returns the highest key and associated container. Last() (key uint64, c *Container) @@ -385,13 +382,8 @@ func (b *Bitmap) remove(v uint64) bool { // Min returns the lowest value in the bitmap. // Second return value is true if containers exist in the bitmap. func (b *Bitmap) Min() (uint64, bool) { - if b.Containers.Size() == 0 { - return 0, false - } - - hb, c := b.Containers.First() - lb, ok := c.min() - return hb<<16 | uint64(lb), ok + v, eof := b.Iterator().Next() + return v, !eof } // Max returns the highest value in the bitmap. @@ -1995,17 +1987,6 @@ func (c *Container) runRemove(v uint16) (*Container, bool) { return c, true } -// min returns the minimum value in the container. -func (c *Container) min() (uint16, bool) { - if c.isArray() { - return c.arrayMin() - } else if c.isRun() { - return c.runMin() - } else { - return c.bitmapMin() - } -} - // max returns the maximum value in the container. func (c *Container) max() uint16 { if c == nil || c.N() == 0 { @@ -2021,34 +2002,11 @@ func (c *Container) max() uint16 { } } -// Second result value is true if array is non-empty. -func (c *Container) arrayMin() (uint16, bool) { - array := c.array() - if len(array) == 0 { - return 0, false - } - return array[0], true -} - func (c *Container) arrayMax() uint16 { array := c.array() return array[len(array)-1] } -// Second result value is true if array is non-empty. -func (c *Container) bitmapMin() (uint16, bool) { - bitmap := c.bitmap() - for i := 0; i < len(bitmap); i++ { - // If value is zero then skip. - v := bitmap[i] - if v != 0 { - r := bits.TrailingZeros64(v) - return uint16(r + i*64), true - } - } - return 0, false -} - func (c *Container) bitmapMax() uint16 { // Search bitmap in reverse order. bitmap := c.bitmap() @@ -2063,15 +2021,6 @@ func (c *Container) bitmapMax() uint16 { return 0 } -// Second result value is true if array is non-empty. -func (c *Container) runMin() (uint16, bool) { - runs := c.runs() - if len(runs) == 0 { - return 0, false - } - return runs[0].start, true -} - func (c *Container) runMax() uint16 { runs := c.runs() if len(runs) == 0 { From 06aa2cf98e455b76222e2324479e53ff2cabe861 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Sat, 15 Jun 2019 15:15:07 +0300 Subject: [PATCH 10/11] updated for feedback from PR 1983 --- fragment.go | 27 +++++++++++++-------------- roaring/roaring.go | 6 +++--- 2 files changed, 16 insertions(+), 17 deletions(-) diff --git a/fragment.go b/fragment.go index c6cc03291..728fe9dee 100644 --- a/fragment.go +++ b/fragment.go @@ -118,8 +118,6 @@ type fragment struct { // Stats reporting. maxRowID uint64 - minRowID uint64 - hasRowID bool // Cache containing full rows (not just counts). rowCache bitmapCache @@ -191,9 +189,6 @@ func (f *fragment) Open() error { // Read last bit to determine max row. f.maxRowID = f.storage.Max() / ShardWidth - min, ok := f.storage.Min() - f.minRowID = min / ShardWidth - f.hasRowID = ok f.stats.Gauge("rows", float64(f.maxRowID), 1.0) return nil }(); err != nil { @@ -525,10 +520,6 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err f.maxRowID = rowID f.stats.Gauge("rows", float64(f.maxRowID), 1.0) } - if !f.hasRowID || rowID < f.minRowID { - f.minRowID = rowID - f.hasRowID = true - } return changed, nil } @@ -1040,12 +1031,13 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin // if filter is nil, it returns fragment.minRowID, 1 // if fragment has no rows, it returns 0, 0 func (f *fragment) minRow(filter *Row) (uint64, uint64) { - if f.hasRowID { + minRowID, hasRowID := f.minRowID() + if hasRowID { if filter == nil { - return f.minRowID, 1 + return minRowID, 1 } // iterate from min row ID and return the first that intersects with filter. - for i := f.minRowID; i <= f.maxRowID; i++ { + for i := minRowID; i <= f.maxRowID; i++ { row := f.row(i).Intersect(filter) count := row.Count() if count > 0 { @@ -1060,12 +1052,14 @@ func (f *fragment) minRow(filter *Row) (uint64, uint64) { // if filter is nil, it returns fragment.maxRowID, 1 // if fragment has no rows, it returns 0, 0 func (f *fragment) maxRow(filter *Row) (uint64, uint64) { - if f.hasRowID { + minRowID, hasRowID := f.minRowID() + if hasRowID { if filter == nil { return f.maxRowID, 1 } // iterate back from max row ID and return the first that intersects with filter. - for i := f.maxRowID; i >= f.minRowID; i-- { + // TODO: implement reverse container iteration to improve performance here for sparse data. --Jaffee + for i := f.maxRowID; i >= minRowID; i-- { row := f.row(i).Intersect(filter) count := row.Count() if count > 0 { @@ -2419,6 +2413,11 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error { return nil } +func (f *fragment) minRowID() (uint64, bool) { + min, ok := f.storage.Min() + return min / ShardWidth, ok +} + // rowFilter is a function signature for controlling iteration over containers // in a fragment. It will be invoked on each container found and returns two // booleans. The first is whether the row this container is in should be diff --git a/roaring/roaring.go b/roaring/roaring.go index 7addfc2f4..f7c3feb37 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -2008,12 +2008,12 @@ func (c *Container) arrayMax() uint16 { func (c *Container) bitmapMax() uint16 { // Search bitmap in reverse order. bitmap := c.bitmap() - for i := len(bitmap) - 1; i > 0; i-- { + for i := len(bitmap); i > 0; i-- { // If value is zero then skip. - v := bitmap[i] + v := bitmap[i-1] if v != 0 { r := bits.LeadingZeros64(v) - return uint16(i*64 + 63 - r) + return uint16((i-1)*64 + 63 - r) } } return 0 From 5c59449ed2326ac8dc7d915260bbd1055079bd0d Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Mon, 17 Jun 2019 16:12:43 +0300 Subject: [PATCH 11/11] reset roaring.go and added bitmap.Min --- roaring/roaring.go | 1 + 1 file changed, 1 insertion(+) diff --git a/roaring/roaring.go b/roaring/roaring.go index f7c3feb37..7ac6671b8 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -2015,6 +2015,7 @@ func (c *Container) bitmapMax() uint16 { r := bits.LeadingZeros64(v) return uint16((i-1)*64 + 63 - r) } + } return 0 }