[query] fix a bug when building histograms

src/query/functions/linear/histogram_quantile.go

@@ -133,13 +133,15 @@ func (b indexedBuckets) Less(i, j int) bool {
 
 type bucketedSeries map[string]indexedBuckets
 
-func gatherSeriesToBuckets(metas []block.SeriesMeta) bucketedSeries {
+type validSeriesBuckets []indexedBuckets
+
+func gatherSeriesToBuckets(metas []block.SeriesMeta) validSeriesBuckets {
 	bucketsForID := make(bucketedSeries, initIndexBucketLength)
 	for i, meta := range metas {
 		tags := meta.Tags
 		value, found := tags.Bucket()
 		if !found {
-			// This series does not have a bucket tag; drop it from the output.
+			// this series does not have a bucket tag; drop it from the output.
 			continue
 		}
 
@@ -158,7 +160,7 @@ func gatherSeriesToBuckets(metas []block.SeriesMeta) bucketedSeries {
 		}
 
 		if buckets, found := bucketsForID[string(id)]; !found {
-			// Add a single indexed bucket for this ID with the current index only.
+			// add a single indexed bucket for this ID with the current index only.
 			newBuckets := make([]indexedBucket, 0, initIndexBucketLength)
 			newBuckets = append(newBuckets, newBucket)
 			bucketsForID[string(id)] = indexedBuckets{
@@ -171,23 +173,28 @@ func gatherSeriesToBuckets(metas []block.SeriesMeta) bucketedSeries {
 		}
 	}
 
-	return bucketsForID
+	return sanitizeBuckets(bucketsForID)
 }
 
 // sanitize sorts the bucket maps by upper bound, dropping any series which
 // have less than two buckets, or any that do not have an upper bound of +Inf
-func sanitizeBuckets(bucketMap bucketedSeries) {
-	for k, buckets := range bucketMap {
+func sanitizeBuckets(bucketMap bucketedSeries) validSeriesBuckets {
+	validSeriesBuckets := make(validSeriesBuckets, 0, len(bucketMap))
+	for _, buckets := range bucketMap {
 		if len(buckets.buckets) < 2 {
-			delete(bucketMap, k)
+			continue
 		}
 
 		sort.Sort(buckets)
 		maxBound := buckets.buckets[len(buckets.buckets)-1].upperBound
 		if !math.IsInf(maxBound, 1) {
-			delete(bucketMap, k)
+			continue
 		}
+
+		validSeriesBuckets = append(validSeriesBuckets, buckets)
 	}
+
+	return validSeriesBuckets
 }
 
 func bucketQuantile(q float64, buckets []bucketValue) float64 {
@@ -257,34 +264,30 @@ func (n *histogramQuantileNode) ProcessBlock(
 
 	meta := b.Meta()
 	seriesMetas := utils.FlattenMetadata(meta, stepIter.SeriesMeta())
-	bucketedSeries := gatherSeriesToBuckets(seriesMetas)
+	seriesBuckets := gatherSeriesToBuckets(seriesMetas)
 
 	q := n.op.q
 	if q < 0 || q > 1 {
-		return processInvalidQuantile(queryCtx, q, bucketedSeries, meta, stepIter, n.controller)
+		return processInvalidQuantile(queryCtx, q, seriesBuckets, meta, stepIter, n.controller)
 	}
 
-	return processValidQuantile(queryCtx, q, bucketedSeries, meta, stepIter, n.controller)
+	return processValidQuantile(queryCtx, q, seriesBuckets, meta, stepIter, n.controller)
 }
 
 func setupBuilder(
 	queryCtx *models.QueryContext,
-	bucketedSeries bucketedSeries,
+	seriesBuckets validSeriesBuckets,
 	meta block.Metadata,
 	stepIter block.StepIter,
 	controller *transform.Controller,
 ) (block.Builder, error) {
-	metas := make([]block.SeriesMeta, len(bucketedSeries))
-	idx := 0
-	for _, v := range bucketedSeries {
-		metas[idx] = block.SeriesMeta{
+	metas := make([]block.SeriesMeta, 0, len(seriesBuckets))
+	for _, v := range seriesBuckets {
+		metas = append(metas, block.SeriesMeta{
 			Tags: v.tags,
-		}
-
-		idx++
+		})
 	}
 
-	meta.Tags, metas = utils.DedupeMetadata(metas, meta.Tags.Opts)
 	builder, err := controller.BlockBuilder(queryCtx, meta, metas)
 	if err != nil {
 		return nil, err
@@ -300,14 +303,12 @@ func setupBuilder(
 func processValidQuantile(
 	queryCtx *models.QueryContext,
 	q float64,
-	bucketedSeries bucketedSeries,
+	seriesBuckets validSeriesBuckets,
 	meta block.Metadata,
 	stepIter block.StepIter,
 	controller *transform.Controller,
 ) (block.Block, error) {
-	sanitizeBuckets(bucketedSeries)
-
-	builder, err := setupBuilder(queryCtx, bucketedSeries, meta, stepIter, controller)
+	builder, err := setupBuilder(queryCtx, seriesBuckets, meta, stepIter, controller)
 	if err != nil {
 		return nil, err
 	}
@@ -317,9 +318,8 @@ func processValidQuantile(
 		values := step.Values()
 		bucketValues := make([]bucketValue, 0, initIndexBucketLength)
 
-		aggregatedValues := make([]float64, len(bucketedSeries))
-		idx := 0
-		for _, b := range bucketedSeries {
+		aggregatedValues := make([]float64, 0, len(seriesBuckets))
+		for _, b := range seriesBuckets {
 			buckets := b.buckets
 			// clear previous bucket values.
 			bucketValues = bucketValues[:0]
@@ -336,8 +336,7 @@ func processValidQuantile(
 				}
 			}
 
-			aggregatedValues[idx] = bucketQuantile(q, bucketValues)
-			idx++
+			aggregatedValues = append(aggregatedValues, bucketQuantile(q, bucketValues))
 		}
 
 		if err := builder.AppendValues(index, aggregatedValues); err != nil {
@@ -355,12 +354,12 @@ func processValidQuantile(
 func processInvalidQuantile(
 	queryCtx *models.QueryContext,
 	q float64,
-	bucketedSeries bucketedSeries,
+	seriesBuckets validSeriesBuckets,
 	meta block.Metadata,
 	stepIter block.StepIter,
 	controller *transform.Controller,
 ) (block.Block, error) {
-	builder, err := setupBuilder(queryCtx, bucketedSeries, meta, stepIter, controller)
+	builder, err := setupBuilder(queryCtx, seriesBuckets, meta, stepIter, controller)
 	if err != nil {
 		return nil, err
 	}
@@ -373,7 +372,7 @@ func processInvalidQuantile(
 	}
 
 	setValue := math.Inf(sign)
-	outValues := make([]float64, len(bucketedSeries))
+	outValues := make([]float64, len(seriesBuckets))
 	util.Memset(outValues, setValue)
 	for index := 0; stepIter.Next(); index++ {
 		if err := builder.AppendValues(index, outValues); err != nil {

src/query/functions/linear/histogram_quantile_test.go

@@ -92,7 +92,7 @@ func TestGatherSeriesToBuckets(t *testing.T) {
 		},
 	}
 
-	assert.Equal(t, expected, actual)
+	assert.Equal(t, sanitizeBuckets(expected), actual)
 }
 
 func TestSanitizeBuckets(t *testing.T) {
@@ -132,8 +132,8 @@ func TestSanitizeBuckets(t *testing.T) {
 		},
 	}
 
-	actual := bucketedSeries{
-		`{bar="baz"}`: indexedBuckets{
+	expected := validSeriesBuckets{
+		indexedBuckets{
 			buckets: []indexedBucket{
 				{upperBound: 1, idx: 0},
 				{upperBound: 2, idx: 3},
@@ -143,8 +143,7 @@ func TestSanitizeBuckets(t *testing.T) {
 		},
 	}
 
-	sanitizeBuckets(bucketed)
-	assert.Equal(t, actual, bucketed)
+	assert.Equal(t, expected, sanitizeBuckets(bucketed))
 }
 
 func TestBucketQuantile(t *testing.T) {