Implement timestamp() for native histograms

Signed-off-by: 🌲 Harry 🌊 John 🏔 <[email protected]>

Author: harry671003

engine/engine_test.go

@@ -2330,6 +2330,16 @@ or
 			end:   time.UnixMilli(160000),
 			step:  time.Minute + 16*time.Second,
 		},
+		{
+			name: "native histogram timestamp",
+			load: `load 2m
+			    http_request_duration_seconds{pod="nginx-1"} {{schema:0 count:3 sum:14.00 buckets:[1 2]}}+{{schema:0 count:4 buckets:[1 2 1]}}x20
+			    http_request_duration_seconds{pod="nginx-2"} 1x10 {{schema:0 count:2 sum:14.00 buckets:[2]}}+{{schema:0 count:6 buckets:[2 2 2]}}x10`,
+			query: `--timestamp(-{__name__="http_request_duration_seconds"} offset 8s)`,
+			start: time.UnixMilli(0),
+			end:   time.UnixMilli(300000),
+			step:  15 * time.Second,
+		},
 	}
 
 	disableOptimizerOpts := []bool{true, false}

engine/testdata/fuzz/FuzzNativeHistogramQuery/d17ae2e47e8e700d

@@ -0,0 +1,10 @@
+go test fuzz v1
+int64(111)
+uint32(0)
+uint32(151)
+uint32(3)
+int8(0)
+int8(0)
+uint64(15)
+uint64(0)
+uint64(106)

execution/function/operator.go

@@ -27,7 +27,7 @@ func NewFunctionOperator(funcExpr *logicalplan.FunctionCall, nextOps []model.Vec
 	case "scalar":
 		return newScalarOperator(model.NewVectorPoolWithSize(stepsBatch, 1), nextOps[0], opts), nil
 	case "timestamp":
-		return newTimestampOperator(nextOps[0], opts), nil
+		return newTimestampOperator(model.NewVectorPool(stepsBatch), nextOps[0], opts), nil
 	case "label_join", "label_replace":
 		return newRelabelOperator(nextOps[0], funcExpr, opts), nil
 	case "absent":

execution/function/timestamp.go

@@ -16,14 +16,16 @@ import (
 )
 
 type timestampOperator struct {
+	pool *model.VectorPool
 	next model.VectorOperator
 
 	series []labels.Labels
 	once   sync.Once
 }
 
-func newTimestampOperator(next model.VectorOperator, opts *query.Options) model.VectorOperator {
+func newTimestampOperator(pool *model.VectorPool, next model.VectorOperator, opts *query.Options) model.VectorOperator {
 	oper := &timestampOperator{
+		pool: pool,
 		next: next,
 	}
 	return telemetry.NewOperator(telemetry.NewTelemetry(oper, opts), oper)
@@ -65,7 +67,7 @@ func (o *timestampOperator) loadSeries(ctx context.Context) error {
 }
 
 func (o *timestampOperator) GetPool() *model.VectorPool {
-	return o.next.GetPool()
+	return o.pool
 }
 
 func (o *timestampOperator) Next(ctx context.Context) ([]model.StepVector, error) {
@@ -76,13 +78,29 @@ func (o *timestampOperator) Next(ctx context.Context) ([]model.StepVector, error
 	}
 
 	in, err := o.next.Next(ctx)
+
 	if err != nil {
 		return nil, err
 	}
+	if len(in) == 0 {
+		return nil, nil
+	}
+	result := o.pool.GetVectorBatch()
 	for _, vector := range in {
+		out := o.pool.GetStepVector(vector.T)
+		value := float64(vector.T / 1000)
+
 		for i := range vector.Samples {
-			vector.Samples[i] = float64(vector.T / 1000)
+			out.AppendSample(o.pool, vector.SampleIDs[i], value)
 		}
+		for i := range vector.Histograms {
+			out.AppendSample(o.pool, vector.HistogramIDs[i], value)
+		}
+
+		result = append(result, out)
+		o.next.GetPool().PutStepVector(vector)
 	}
-	return in, nil
+
+	o.next.GetPool().PutVectors(in)
+	return result, nil
 }