1 files changed, 18 insertions, 4 deletions
diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go
index c50c7a1..3281353 100644
--- a/internal/mapr/aggregateset.go
+++ b/internal/mapr/aggregateset.go
@@ -46,6 +46,10 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 		case Sum:
 			fallthrough
 		case Avg:
+			fallthrough
+		case Percentage:
+			fallthrough
+		case Percentile:
 			value := set.FValues[storage]
 			s.addFloat(storage, value)
 		case Min:
@@ -67,21 +71,25 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error {
 	return nil
 }
 
-// Serialize the aggregate set so it can be sent over the wire.
-func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) {
+// Serialize the aggregate set so it can be sent over the wire. Returns true
+// when the serialized message was successfully sent, and false when the
+// context was cancelled before the send completed. Callers that own the
+// source state (e.g. Aggregate) must re-merge unsent sets so data is
+// not silently lost.
+func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) bool {
 	dlog.Common.Trace("Serialising mapr.AggregateSet", s)
 	sb := pool.BuilderBuffer.Get().(*strings.Builder)
 	defer pool.RecycleBuilderBuffer(sb)
 
 	sb.WriteString(groupKey)
 	sb.WriteString(protocol.AggregateDelimiter)
-	sb.WriteString(fmt.Sprintf("%d", s.Samples))
+	sb.WriteString(strconv.Itoa(s.Samples))
 	sb.WriteString(protocol.AggregateDelimiter)
 
 	for k, v := range s.FValues {
 		sb.WriteString(k)
 		sb.WriteString(protocol.AggregateKVDelimiter)
-		sb.WriteString(fmt.Sprintf("%v", v))
+		sb.WriteString(strconv.FormatFloat(v, 'f', -1, 64))
 		sb.WriteString(protocol.AggregateDelimiter)
 	}
 
@@ -94,7 +102,9 @@ func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<-
 
 	select {
 	case ch <- sb.String():
+		return true
 	case <-ctx.Done():
+		return false
 	}
 }
 
@@ -177,6 +187,10 @@ func (s *AggregateSet) Aggregate(key string, agg AggregateOperation, value strin
 	case Sum:
 		fallthrough
 	case Avg:
+		fallthrough
+	case Percentage:
+		fallthrough
+	case Percentile:
 		s.addFloat(key, f)
 	case Min:
 		s.addFloatMin(key, f)