add support for binaryOps for native histograms (#526)

* support binary OR between native histograms

Signed-off-by: Saumya Shah <[email protected]>

* add test

Signed-off-by: Saumya Shah <[email protected]>

* update tests

Signed-off-by: Saumya Shah <[email protected]>

* fix, consider all four cases while doing OR

Signed-off-by: Saumya Shah <[email protected]>

* fix: consider all cases with proper alignments

Signed-off-by: Saumya Shah <[email protected]>

* and and unless for native histograms

Signed-off-by: Saumya Shah <[email protected]>

* lint

Signed-off-by: Saumya Shah <[email protected]>

* add variations between float and native histograms in many-to-many joins in tests

Signed-off-by: Saumya Shah <[email protected]>

* format

Signed-off-by: Saumya Shah <[email protected]>

---------

Signed-off-by: Saumya Shah <[email protected]>

Author: Saumya40-codes

engine/engine_test.go

@@ -5250,7 +5250,6 @@ func TestNativeHistograms(t *testing.T) {
 			name:  "rate()",
 			query: `rate(native_histogram_series[1m])`,
 		},
-
 		{
 			name:  "increase()",
 			query: `increase(native_histogram_series[1m])`,
@@ -5375,6 +5374,36 @@ histogram_sum(
 			name:  "subqueries",
 			query: `increase(rate(native_histogram_series[2m])[2m:15s])`,
 		},
+		{
+			name: "Binary OR",
+			query: `
+  native_histogram_series
+or
+  (histogram_quantile(0.7, native_histogram_series) or rate(native_histogram_series[2m]))`,
+		},
+		{
+			name:  "Mixed Binary OR",
+			query: `sum(native_histogram_series) or native_histogram_series`, // sum will be a single float value, float series on lhs of 'or'
+		},
+		{
+			name: "Binary AND",
+			query: `
+  (rate(native_histogram_series[2m]) and histogram_quantile(0.7, native_histogram_series))
+and
+  native_histogram_series`,
+		},
+		{
+			name:  "Mixed Binary AND",
+			query: `native_histogram_series and count(native_histogram_series)`, // count will be a single float value, float series on 'rhs' of 'and'
+		},
+		{
+			name:  "many-to-many join Unless",
+			query: `sum without (foo) (native_histogram_series) unless native_histogram_series / 2`,
+		},
+		{
+			name:  "Mixed many-to-many join Unless",
+			query: `native_histogram_series * 3 unless avg(native_histogram_series)`,
+		},
 	}
 
 	defer pprof.StopCPUProfile()
@@ -5485,6 +5514,7 @@ func generateNativeHistogramSeries(app storage.Appender, numSeries int, withMixe
 		PositiveBuckets: []int64{1, 2, -2, 1, -1, 0, 3},
 		Count:           13,
 	}
+
 	for sid, histograms := range series {
 		lbls := append(commonLabels, "h", strconv.Itoa(sid))
 		for i := range histograms {

execution/binary/vector.go

@@ -232,17 +232,19 @@ func (o *vectorOperator) execBinaryAnd(lhs, rhs model.StepVector) (model.StepVec
 		jp.sid = sampleID
 		jp.ats = ts
 	}
-	for i, sampleID := range lhs.SampleIDs {
-		if jp := o.hcJoinBuckets[sampleID]; jp.ats == ts {
-			step.AppendSample(o.pool, o.outputSeriesID(sampleID+1, jp.sid+1), lhs.Samples[i])
-		}
-	}
 
 	for _, histogramID := range rhs.HistogramIDs {
 		jp := o.lcJoinBuckets[histogramID]
 		jp.sid = histogramID
 		jp.ats = ts
 	}
+
+	for i, sampleID := range lhs.SampleIDs {
+		if jp := o.hcJoinBuckets[sampleID]; jp.ats == ts {
+			step.AppendSample(o.pool, o.outputSeriesID(sampleID+1, jp.sid+1), lhs.Samples[i])
+		}
+	}
+
 	for i, histogramID := range lhs.HistogramIDs {
 		if jp := o.hcJoinBuckets[histogramID]; jp.ats == ts {
 			step.AppendHistogram(o.pool, o.outputSeriesID(histogramID+1, jp.sid+1), lhs.Histograms[i])
@@ -260,11 +262,25 @@ func (o *vectorOperator) execBinaryOr(lhs, rhs model.StepVector) (model.StepVect
 		jp.ats = ts
 		step.AppendSample(o.pool, o.outputSeriesID(sampleID+1, 0), lhs.Samples[i])
 	}
+
+	for i, histogramID := range lhs.HistogramIDs {
+		jp := o.hcJoinBuckets[histogramID]
+		jp.ats = ts
+		step.AppendHistogram(o.pool, o.outputSeriesID(histogramID+1, 0), lhs.Histograms[i])
+	}
+
 	for i, sampleID := range rhs.SampleIDs {
 		if jp := o.lcJoinBuckets[sampleID]; jp.ats != ts {
 			step.AppendSample(o.pool, o.outputSeriesID(0, sampleID+1), rhs.Samples[i])
 		}
 	}
+
+	for i, histogramID := range rhs.HistogramIDs {
+		if jp := o.lcJoinBuckets[histogramID]; jp.ats != ts {
+			step.AppendHistogram(o.pool, o.outputSeriesID(0, histogramID+1), rhs.Histograms[i])
+		}
+	}
+
 	return step, nil
 }
 
@@ -276,11 +292,21 @@ func (o *vectorOperator) execBinaryUnless(lhs, rhs model.StepVector) (model.Step
 		jp := o.lcJoinBuckets[sampleID]
 		jp.ats = ts
 	}
+	for _, histogramID := range rhs.HistogramIDs {
+		jp := o.lcJoinBuckets[histogramID]
+		jp.ats = ts
+	}
+
 	for i, sampleID := range lhs.SampleIDs {
 		if jp := o.hcJoinBuckets[sampleID]; jp.ats != ts {
 			step.AppendSample(o.pool, o.outputSeriesID(sampleID+1, 0), lhs.Samples[i])
 		}
 	}
+	for i, histogramID := range lhs.HistogramIDs {
+		if jp := o.hcJoinBuckets[histogramID]; jp.ats != ts {
+			step.AppendHistogram(o.pool, o.outputSeriesID(histogramID+1, 0), lhs.Histograms[i])
+		}
+	}
 	return step, nil
 }