Merge pull request #1483 from jaffee/core-101-disco

CORE-101 Percentiles on disco

Makefile

@@ -236,7 +236,7 @@ topt:
 
 topt-race:
 	mv log.topt.race log.topt.race.prev || true
-	$(eval SHELL:=/bin/bash) set -o pipefail; CGO_ENABLED=1 go test -race -v -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.race
+	$(eval SHELL:=/bin/bash) set -o pipefail; CGO_ENABLED=1 go test -race -timeout 60m -v -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.race
 	@echo "   log.topt.race green: \c"; cat log.topt.race | grep PASS |wc -l
 	@echo "   log.topt.race   red: \c"; cat log.topt.race | grep '\-\-\- FAIL' | wc -l
 

executor.go

@@ -830,6 +830,9 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p
 	case "Limit":
 		res, err := e.executeLimitCall(ctx, qcx, index, c, shards, opt)
 		return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards))
+	case "Percentile":
+		res, err := e.executePercentile(ctx, qcx, index, c, shards, opt)
+		return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards))
 	default: // e.g. "Row", "Union", "Intersect" or anything that returns a bitmap.
 		statFn()
 		res, err := e.executeBitmapCall(ctx, qcx, index, c, shards, opt)
@@ -1293,6 +1296,119 @@ func (e *executor) executeMax(ctx context.Context, qcx *Qcx, index string, c *pq
 	return other, nil
 }
 
+// executePercentile executes a Percentile() call.
+func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (_ ValCount, err error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executePercentile")
+	defer span.Finish()
+
+	// get nth
+	var nth float64
+	if nthArg, ok := c.Args["nth"].(pql.Decimal); ok {
+		nth = nthArg.Float64()
+		if nth < 0 || nth > 1.0 {
+			return ValCount{}, errors.Errorf("Percentile(): invalid nth value(%f), should be >= 0 and <= 1.0", nth)
+		}
+	} else {
+		return ValCount{}, errors.New("Percentile(): nth required")
+	}
+
+	// get field
+	if fieldArg := c.Args["field"]; fieldArg == "" {
+		return ValCount{}, errors.New("Percentile(): field required")
+	}
+	fieldName, _, _ := c.StringArg("field")
+
+	// filter call for min & max
+	var filterCall *pql.Call
+
+	// check if filter provided
+	if filterArg, ok := c.Args["filter"].(*pql.Call); ok && filterArg != nil {
+		filterCall = filterArg
+	}
+
+	// get min
+	q, _ := pql.ParseString(fmt.Sprintf(`Min(field="%s")`, fieldName))
+	minCall := q.Calls[0]
+	if filterCall != nil {
+		minCall.Children = append(minCall.Children, filterCall)
+	}
+	minVal, err := e.executeMin(ctx, qcx, index, minCall, shards, opt)
+	if err != nil {
+		return ValCount{}, errors.Wrap(err, "executing Min call for Percentile")
+	}
+	if nth == 0.0 {
+		return ValCount{Val: minVal.Val, Count: minVal.Count}, nil
+	}
+
+	// get max
+	q, _ = pql.ParseString(fmt.Sprintf(`Max(field="%s")`, fieldName))
+	maxCall := q.Calls[0]
+	if filterCall != nil {
+		maxCall.Children = append(maxCall.Children, filterCall)
+	}
+	maxVal, err := e.executeMax(ctx, qcx, index, maxCall, shards, opt)
+	if err != nil {
+		return ValCount{}, errors.Wrap(err, "executing Max call for Percentile")
+	}
+	// set up reusables
+	var countCall, rangeCall *pql.Call
+	if filterCall == nil {
+		countQuery, _ := pql.ParseString(fmt.Sprintf("Count(Row(%s < 0))", fieldName))
+		countCall = countQuery.Calls[0]
+		rangeCall = countCall.Children[0]
+	} else {
+		countQuery, _ := pql.ParseString(fmt.Sprintf(`Count(Intersect(Row(%s < 0)))`, fieldName))
+		countCall = countQuery.Calls[0]
+		intersectCall := countCall.Children[0]
+		intersectCall.Children = append(intersectCall.Children, filterCall)
+		rangeCall = intersectCall.Children[0]
+	}
+
+	k := (1 - nth) / nth
+
+	min, max := minVal.Val, maxVal.Val
+	// estimate nth val, eg median when nth=0.5
+	for min < max {
+		possibleNthVal := (max + min) / 2
+		// get left count
+		rangeCall.Args[fieldName] = &pql.Condition{
+			Op:    pql.Token(pql.LT),
+			Value: possibleNthVal,
+		}
+		leftCountUint64, err := e.executeCount(ctx, qcx, index, countCall, shards, opt)
+		if err != nil {
+			return ValCount{}, errors.Wrap(err, "executing Count call L for Percentile")
+		}
+		leftCount := int64(leftCountUint64)
+
+		// get right count
+		rangeCall.Args[fieldName] = &pql.Condition{
+			Op:    pql.Token(pql.GT),
+			Value: possibleNthVal,
+		}
+		rightCountUint64, err := e.executeCount(ctx, qcx, index, countCall, shards, opt)
+		if err != nil {
+			return ValCount{}, errors.Wrap(err, "executing Count call R for Percentile")
+		}
+		rightCount := int64(rightCountUint64)
+
+		// 'weight' the left count as per k
+		leftCountWeighted := int64(math.Round(k * float64(leftCount)))
+
+		// binary search
+		if leftCountWeighted > rightCount {
+			max = possibleNthVal - 1
+		} else if leftCountWeighted < rightCount {
+			min = possibleNthVal + 1
+		} else {
+			return ValCount{Val: possibleNthVal, Count: 1}, nil
+		}
+	}
+
+	return ValCount{Val: min, Count: 1}, nil
+
+}
+
 // executeMinRow executes a MinRow() call.
 func (e *executor) executeMinRow(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (_ interface{}, err error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRow")

executor_test.go

@@ -25,6 +25,7 @@ import (
 	"io/ioutil"
 	"math"
 	"math/rand"
+	_ "net/http/pprof"
 	"reflect"
 	"sort"
 	"strconv"
@@ -6850,12 +6851,178 @@ func TestVariousQueries(t *testing.T) {
 	for _, clusterSize := range []int{1, 3, 4, 7} {
 		clusterSize := clusterSize
 		t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) {
-			t.Parallel()
 			c := test.MustRunCluster(t, clusterSize)
 			defer c.Close()
 
 			variousQueries(t, c)
 			variousQueriesOnTimeFields(t, c)
+			variousQueriesOnPercentiles(t, c)
+		})
+	}
+}
+
+// tests for abbreviating time values in queries
+func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) {
+	// todo, make rand more random, 42 isnt the answer to everything
+	// however, to make tests reproducible, seed should be printed
+	// on failure?
+	r := rand.New(rand.NewSource(42))
+
+	// gen Numbers to test percentile query on, shuffle for extra spice
+	// size should always be greater than 0
+	type testValue struct {
+		colKey string
+		num    int64
+		rowKey string
+	}
+	size := 100
+
+	testValues := make([]testValue, size)
+	rowKeys := [2]string{"foo", "bar"}
+	for i := 0; i < size; i++ {
+		num := int64(r.Uint32())
+		// flip coin to negate
+		if r.Uint64()%2 == 0 {
+			num = -num
+		}
+		testValues[i] = testValue{
+			colKey: fmt.Sprintf("user%d", i+1),
+			num:    num,
+			rowKey: rowKeys[r.Uint64()%2], // flip a coin
+		}
+	}
+
+	// filter out nums that fulfil predicate
+	var nums []int64
+	for _, v := range testValues {
+		if v.rowKey == "foo" {
+			nums = append(nums, v.num)
+		}
+	}
+
+	// get min and max for calculating both expected median
+	// and bounds for bsi field
+	// get min & max
+
+	// helper function for calculating percentiles to
+	// cross-check with Pilosa's results
+	getExpectedPercentile := func(nums []int64, nth float64) int64 {
+		min, max := nums[0], nums[0]
+		for _, num := range nums {
+			if num < min {
+				min = num
+			}
+			if num > max {
+				max = num
+			}
+		}
+		if nth == 0.0 {
+			return min
+		}
+		k := (1 - nth) / nth
+
+		possibleNthVal := int64(0)
+		// bin search
+		for min < max {
+			possibleNthVal = (max + min) / 2
+			leftCount, rightCount := int64(0), int64(0)
+			for _, num := range nums {
+				if num < possibleNthVal {
+					leftCount++
+				} else if num > possibleNthVal {
+					rightCount++
+				}
+			}
+
+			leftCountWeighted := int64(math.Round(k * float64(leftCount)))
+
+			if leftCountWeighted > rightCount {
+				max = possibleNthVal - 1
+			} else if leftCountWeighted < rightCount {
+				min = possibleNthVal + 1
+			} else { // perfectly balanced, as all things should be
+				return possibleNthVal
+			}
+		}
+		return min
+	}
+
+	// generate numeric entries for index
+	intEntries := make([]test.IntKey, size)
+	for i := 0; i < size; i++ {
+		key := testValues[i].colKey
+		val := testValues[i].num
+		intEntries[i] = test.IntKey{Key: key, Val: val}
+	}
+
+	// generate string-set entries for index
+	var stringEntries [][2]string
+	for _, v := range testValues {
+		stringEntries = append(stringEntries,
+			[2]string{v.rowKey, v.colKey})
+	}
+
+	// get min max for bsi bounds
+	min, max := testValues[0].num, testValues[0].num
+	for _, v := range testValues {
+		if v.num < min {
+			min = v.num
+		}
+		if v.num > max {
+			max = v.num
+		}
+	}
+
+	// generic index
+	c.CreateField(t, "users2", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(min, max))
+	c.ImportIntKey(t, "users2", "net_worth", intEntries)
+
+	c.CreateField(t, "users2", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "val", pilosa.OptFieldKeys())
+	c.ImportKeyKey(t, "users2", "val", stringEntries)
+
+	splitSortBackToCSV := func(csvStr string) string {
+		ss := strings.Split(csvStr[:len(csvStr)-1], "\n")
+		sort.Strings(ss)
+		return strings.Join(ss, "\n") + "\n"
+	}
+
+	type testCase struct {
+		query string
+		// qrVerifier  func(t *testing.T, resp pilosa.QueryResponse)
+		csvVerifier string
+	}
+
+	// generate test cases per each nth argument
+	nths := []float64{0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99}
+	var tests []testCase
+	for _, nth := range nths {
+		query := fmt.Sprintf(`Percentile(field="net_worth", filter=Row(val="foo"), nth=%f)`, nth)
+		expectedPercentile := getExpectedPercentile(nums, nth)
+		tests = append(tests, testCase{
+			query:       query,
+			csvVerifier: fmt.Sprintf("%d,1\n", expectedPercentile),
+		})
+	}
+
+	for i, tst := range tests {
+		t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) {
+			// resp := c.Query(t, "users2", tst.query)
+			tr := c.QueryGRPC(t, "users2", tst.query)
+			// if tst.qrVerifier != nil {
+			// 	tst.qrVerifier(t, resp)
+			// }
+			csvString, err := tableResponseToCSVString(tr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			// verify everything after header
+			got := splitSortBackToCSV(csvString[strings.Index(csvString, "\n")+1:])
+			if got != tst.csvVerifier {
+				t.Errorf("expected:\n%s\ngot:\n%s", tst.csvVerifier, got)
+			}
+
+			// TODO: add HTTP and Postgres and ability to convert
+			// those results to CSV to run through CSV verifier
 		})
 	}
 }

pql/ast.go

@@ -433,6 +433,14 @@ var callInfoByFunc = map[string]callInfo{
 
 	// things that take _field
 	"TopN": allowUnderField,
+	"Percentile": {
+		allowUnknown: false,
+		prototypes: map[string]interface{}{
+			"field":  "",
+			"filter": nil,
+			"nth":    nil,
+		},
+	},
 	// special cases:
 	"Clear": {
 		allowUnknown: true,

pql/pql.peg

@@ -14,6 +14,7 @@ Call <-  "Set" {p.startCall("Set")} open col comma args (comma timestamp)? close
        / "Store" {p.startCall("Store")} open Call comma arg close {p.endCall()}
        / "TopN" {p.startCall("TopN")} open posfield (comma allargs)? close {p.endCall()}
        / "TopK" {p.startCall("TopK")} open posfield (comma allargs)? close {p.endCall()}
+       / "Percentile" {p.startCall("Percentile")} open posfield (comma allargs)? close {p.endCall()}       
        / "Rows" {p.startCall("Rows")} open posfield (comma allargs)? close {p.endCall()}
        / "Range" {p.startCall("Range")} open field eq value comma 'from='? {p.addField("from")} timestampfmt {p.addVal(text)} comma 'to='? sp {p.addField("to")} timestampfmt {p.addVal(text)} close {p.endCall()}
        / < IDENT > { p.startCall(text) } open allargs comma? close { p.endCall() }