Add percentage and percentile mapr aggregators

1 files changed, 56 insertions, 2 deletions

diff --git a/internal/mapr/groupset.go b/internal/mapr/groupset.go
index 9d7661a..144f2cb 100644
--- a/internal/mapr/groupset.go
+++ b/internal/mapr/groupset.go
@@ -24,6 +24,11 @@ type result struct {
 	orderBy      float64
 }
 
+type resultStats struct {
+	percentageTotals map[string]float64
+	percentileValues map[string][]float64
+}
+
 // NewGroupSet returns a new empty group set.
 func NewGroupSet() *GroupSet {
 	g := GroupSet{}
@@ -67,12 +72,13 @@ func (g *GroupSet) result(query *Query, gathercolumnWidths bool) ([]result, []in
 	// not a CSV file).
 	columnWidths := make([]int, len(query.Select))
 	var valueStrLen int
+	stats := g.makeResultStats(query)
 
 	for groupKey, set := range g.sets {
 		result := result{groupKey: groupKey}
 
 		for i, sc := range query.Select {
-			if valueStrLen, err = g.resultSelect(query, &sc, set, &result); err != nil {
+			if valueStrLen, err = g.resultSelect(query, &sc, set, &result, &stats); err != nil {
 				return rows, columnWidths, err
 			}
 
@@ -96,7 +102,7 @@ func (g *GroupSet) result(query *Query, gathercolumnWidths bool) ([]result, []in
 }
 
 func (*GroupSet) resultSelect(query *Query, sc *selectCondition, set *AggregateSet,
-	result *result) (int, error) {
+	result *result, stats *resultStats) (int, error) {
 
 	var valueStr string
 	var value float64
@@ -120,6 +126,18 @@ func (*GroupSet) resultSelect(query *Query, sc *selectCondition, set *AggregateS
 	case Avg:
 		value = set.FValues[sc.FieldStorage] / float64(set.Samples)
 		valueStr = fmt.Sprintf("%f", value)
+	case Percentage:
+		value = set.FValues[sc.FieldStorage]
+		total := stats.percentageTotals[sc.FieldStorage]
+		if total == 0 {
+			value = 0
+		} else {
+			value = (value / total) * 100
+		}
+		valueStr = fmt.Sprintf("%f", value)
+	case Percentile:
+		value = percentileRank(set.FValues[sc.FieldStorage], stats.percentileValues[sc.FieldStorage])
+		valueStr = fmt.Sprintf("%f", value)
 	default:
 		return 0, fmt.Errorf("Unknown aggregation method '%v'", sc.Operation)
 	}
@@ -132,6 +150,42 @@ func (*GroupSet) resultSelect(query *Query, sc *selectCondition, set *AggregateS
 	return len(valueStr), nil
 }
 
+func (g *GroupSet) makeResultStats(query *Query) resultStats {
+	stats := resultStats{
+		percentageTotals: make(map[string]float64),
+		percentileValues: make(map[string][]float64),
+	}
+
+	for _, set := range g.sets {
+		for _, sc := range query.Select {
+			value := set.FValues[sc.FieldStorage]
+			switch sc.Operation {
+			case Percentage:
+				stats.percentageTotals[sc.FieldStorage] += value
+			case Percentile:
+				stats.percentileValues[sc.FieldStorage] = append(stats.percentileValues[sc.FieldStorage], value)
+			}
+		}
+	}
+
+	for storage := range stats.percentileValues {
+		sort.Float64s(stats.percentileValues[storage])
+	}
+
+	return stats
+}
+
+func percentileRank(value float64, sortedValues []float64) float64 {
+	if len(sortedValues) == 0 {
+		return 0
+	}
+
+	upperBound := sort.Search(len(sortedValues), func(i int) bool {
+		return sortedValues[i] > value
+	})
+	return (float64(upperBound) / float64(len(sortedValues))) * 100
+}
+
 func (*GroupSet) resultOrderBy(query *Query, rows []result) {
 	if query.OrderBy == "" {
 		return