Merge pull request #784 from 44YHC/fixes

#752, #753 Fixes

Author: rita7lopes

reader/model/responses.go

@@ -38,7 +38,7 @@ type TraceInfo struct {
 	RootServiceName   string    `json:"rootServiceName"`
 	RootTraceName     string    `json:"rootTraceName"`
 	StartTimeUnixNano string    `json:"startTimeUnixNano"`
-	DurationMs        float64   `json:"durationMs"`
+	DurationMs        int64     `json:"durationMs"`
 	SpanSet           SpanSet   `json:"spanSet"`
 	SpanSets          []SpanSet `json:"spanSets"`
 }

reader/traceql/traceql_transpiler/clickhouse_transpiler/planner_test.go

@@ -2,6 +2,7 @@ package clickhouse_transpiler
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -44,6 +45,47 @@ func TestPlanner(t *testing.T) {
 	fmt.Println(res)
 }
 
+// TestTracesDataDurationMsSQL verifies that the generated SQL uses intDiv (integer
+// division) for _duration_ms rather than toFloat64, which would produce fractional
+// values that Grafana's Tempo datasource cannot unmarshal into uint32 (issue #782).
+func TestTracesDataDurationMsSQL(t *testing.T) {
+	script, err := traceql_parser.Parse(`{.service.name="test"}`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	plan, err := Plan(script)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req, err := plan.Process(&shared.PlannerContext{
+		IsCluster:            false,
+		From:                 time.Now().Add(time.Hour * -1),
+		To:                   time.Now(),
+		Limit:                10,
+		TracesAttrsTable:     "tempo_traces_attrs_gin",
+		TracesAttrsDistTable: "tempo_traces_attrs_gin_dist",
+		TracesTable:          "tempo_traces",
+		TracesDistTable:      "tempo_traces_dist",
+		VersionInfo:          map[string]int64{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	query, err := req.String(&sql.Ctx{
+		Params: map[string]sql.SQLObject{},
+		Result: map[string]sql.SQLObject{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(query, "toFloat64") {
+		t.Errorf("generated SQL must not contain toFloat64 for _duration_ms; got:\n%s", query)
+	}
+	if !strings.Contains(query, "intDiv") {
+		t.Errorf("generated SQL must use intDiv for _duration_ms; got:\n%s", query)
+	}
+}
+
 func TestComplexPlanner(t *testing.T) {
 	script, err := traceql_parser.Parse(`{.randomContainer=~"admiring" && .randomFloat > 10} | count() > 2 || {.randomContainer=~"boring" && .randomFloat < 10}`)
 	if err != nil {

reader/traceql/traceql_transpiler/clickhouse_transpiler/traces_data.go

@@ -46,7 +46,7 @@ func (t *TracesDataPlanner) Process(ctx *shared.PlannerContext) (sql.ISelect, er
 			Select(
 				sql.NewSimpleCol("traces.trace_id", "trace_id"),
 				sql.NewSimpleCol("min(traces.timestamp_ns)", "_start_time_unix_nano"),
-				sql.NewSimpleCol("toFloat64(max(traces.timestamp_ns + traces.duration_ns) - min(traces.timestamp_ns)) / 1000000", "_duration_ms"),
+				sql.NewSimpleCol("intDiv(max(traces.timestamp_ns + traces.duration_ns) - min(traces.timestamp_ns), 1000000)", "_duration_ms"),
 				sql.NewSimpleCol("argMin(traces.service_name, traces.timestamp_ns)", "_root_service_name"),
 				sql.NewSimpleCol("argMin(traces.name, traces.timestamp_ns)", "_root_trace_name")).
 			From(sql.NewSimpleCol(ctx.TracesTable, "traces")).

reader/traceql/traceql_transpiler/reqest_processor.go

@@ -45,7 +45,7 @@ func (t TraceQLRequestProcessor) Process(ctx *shared.PlannerContext) (chan []mod
 				durationsNs       []int64
 				timestampsNs      []int64
 				startTimeUnixNano int64
-				traceDurationMs   float64
+				traceDurationMs   int64
 				rootServiceName   string
 				rootTraceName     string
 			)

writer/utils/unmarshal/metrics_protobuf.go

@@ -43,7 +43,7 @@ func (l *promMetricsProtoDec) Decode() error {
 			points++
 			if points >= flushLimit {
 				err := l.onEntries(oLblsBuf, tsns, msg, value,
-					fastFillArray[uint8](len(ts.GetSamples()), model.SAMPLE_TYPE_METRIC))
+					fastFillArray[uint8](len(tsns), model.SAMPLE_TYPE_METRIC))
 				if err != nil {
 					return err
 				}

writer/utils/unmarshal/metrics_protobuf_test.go

@@ -0,0 +1,55 @@
+package unmarshal
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/metrico/qryn/v4/writer/utils/proto/prompb"
+)
+
+// TestPromMetricsFlushLimitTypesLength verifies that when a time series has more
+// samples than flushLimit (1000), the types slice passed to onEntries always has
+// the same length as the timestampsNS slice (not the total sample count).
+//
+// This is a regression test for: https://github.com/metrico/gigapipe/issues/783
+func TestPromMetricsFlushLimitTypesLength(t *testing.T) {
+	const totalSamples = 1001 // one more than flushLimit to trigger the mid-flush
+
+	samples := make([]*prompb.Sample, totalSamples)
+	for i := range samples {
+		samples[i] = &prompb.Sample{Timestamp: int64(i), Value: float64(i)}
+	}
+
+	req := &prompb.WriteRequest{
+		Timeseries: []*prompb.TimeSeries{
+			{
+				Labels:  []*prompb.Label{{Name: "__name__", Value: "test_metric"}},
+				Samples: samples,
+			},
+		},
+	}
+
+	dec := &promMetricsProtoDec{
+		ctx: &ParserCtx{bodyObject: req},
+	}
+
+	var mismatches []string
+	dec.SetOnEntries(func(labels [][]string, timestampsNS []int64, message []string, value []float64, types []uint8) error {
+		if len(types) != len(timestampsNS) {
+			mismatches = append(mismatches, fmt.Sprintf(
+				"types length %d != timestampsNS length %d", len(types), len(timestampsNS),
+			))
+		}
+		return nil
+	})
+
+	if err := dec.Decode(); err != nil {
+		t.Fatalf("Decode returned error: %v", err)
+	}
+
+	if len(mismatches) > 0 {
+		for _, m := range mismatches {
+			t.Errorf("column length mismatch: %s", m)
+		}
+	}
+}