diff --git a/cluster.go b/cluster.go index cf0090c0f..c972daadd 100644 --- a/cluster.go +++ b/cluster.go @@ -1702,22 +1702,24 @@ func confirmNodeDown(uri URI, log logger.Logger) bool { Host: uri.HostPort(), Path: "version", } - ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second) - defer cancel() req, err := http.NewRequest("GET", u.String(), nil) if err != nil { log.Printf("bad request:%s %s", u.String(), err) return false } - for i := 0; i < confirmDownRetries; i++ { + ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second) + defer cancel() resp, err := http.DefaultClient.Do(req.WithContext(ctx)) + var bod []byte if err == nil { + bod, err = ioutil.ReadAll(resp.Body) if resp.StatusCode == 200 { return false } } - log.Printf("NodeLeave Timeout with %s %d", uri.HostPort(), i) + + log.Printf("NodeLeave confirm with %s %d. err: '%v' bod: '%s'", uri.HostPort(), i, err, bod) time.Sleep(confirmDownSleep * time.Second) } return true diff --git a/executor.go b/executor.go index 607840a42..7eb13dd79 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") @@ -650,9 +724,6 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal // 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) @@ -689,6 +760,64 @@ 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) { + 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 { + return Pair{}, nil + } + + fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) + if fragment == nil { + return Pair{}, nil + } + + minRowID, count := fragment.minRow(filter) + return Pair{ + ID: minRowID, + Count: count, + }, nil +} + +// 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 { + 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 { + return Pair{}, nil + } + + fragment := e.Holder.fragment(index, fieldName, viewStandard, shard) + if fragment == nil { + return Pair{}, nil + } + + maxRowID, count := fragment.maxRow(filter) + return Pair{ + ID: 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 +2753,25 @@ 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 + } + if call.Name == "MinRow" || call.Name == "MaxRow" { + result.Key = key + return result, nil + } + return Pair{Key: key, Count: result.Count}, nil + } + } + case []Pair: if fieldName := callArgString(call, "_field"); fieldName != "" { field := idx.Field(fieldName) 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) { diff --git a/fragment.go b/fragment.go index daa621ca5..f81a0508a 100644 --- a/fragment.go +++ b/fragment.go @@ -191,8 +191,7 @@ 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 f.stats.Gauge("rows", float64(f.maxRowID), 1.0) return nil }(); err != nil { @@ -1031,6 +1030,49 @@ 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) { + minRowID, hasRowID := f.minRowID() + if hasRowID { + if filter == nil { + return minRowID, 1 + } + // iterate from min row ID and return the first that intersects with filter. + for i := 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) { + 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. + // 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 { + 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 { @@ -2374,6 +2416,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/README.md b/roaring/README.md index 044c98b83..53269cf88 100644 --- a/roaring/README.md +++ b/roaring/README.md @@ -18,7 +18,7 @@ The fuzzer needs some input to start the fuzzing with. Copy some sample Pilosa f Once you have copied your sample inputs, you are ready to run the fuzzer: -`go-fuzz -bin=roaring-fuzz.zip -workdir=workdir -func=FuzzBitmapUnmarshalBianry` +`go-fuzz -bin=roaring-fuzz.zip -workdir=workdir -func=FuzzBitmapUnmarshalBinary` ## Understanding the Fuzzer Output diff --git a/roaring/containers_test.go b/roaring/containers_test.go index abe5dbd99..1869d31af 100644 --- a/roaring/containers_test.go +++ b/roaring/containers_test.go @@ -98,5 +98,4 @@ func testContainersIterator(cs Containers, t *testing.T) { if itr.Next() { t.Fatalf("itr should be done, but got true") } - } diff --git a/roaring/roaring.go b/roaring/roaring.go index bc3fc2e05..b3d26c7a6 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -379,6 +379,13 @@ func (b *Bitmap) remove(v uint64) bool { return changed } +// 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) { + v, eof := b.Iterator().Next() + return v, !eof +} + // Max returns the highest value in the bitmap. // Returns zero if the bitmap is empty. func (b *Bitmap) Max() uint64 { diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index c6ea63eb0..fe966be17 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)