From d2aca3bbfc643d60b2d6accd02a3b575cf4aa1c4 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Fri, 31 May 2019 15:32:15 +0300 Subject: [PATCH] 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=