Merge branch 'master' into tests

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

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)

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) {

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

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
 

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")
 	}
-
 }

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 {

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)