Added MinRow and MaxRow calls

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)

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 {

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=