fix(drilldown): stop injecting unknown_service for non-service volume grouping (#229)

* fix(drilldown): avoid unknown_service volume buckets for non-service grouping

* fix(drilldown): honor fieldBy fallback for volume grouping

* fix(metrics): harden binary scalar handling and add ops matrix

* fix: align rate and unwrap parity with Loki

* fix(translator): keep rate parity for non-stream labels

Author: szibis

CHANGELOG.md

@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Bug Fixes
+
+- drilldown/volume: stop injecting synthetic `service_name="unknown_service"` into `index/volume` and `index/volume_range` buckets when requests are grouped by non-service labels (for example `cluster`), while preserving service-aware grouping behavior.
+- drilldown/volume: honor Drilldown grouping hints (`drillDownLabel`, `fieldBy`, and `var-fieldBy`) as target-label fallbacks when `targetLabels` is omitted, so field/label include-exclude actions keep grouping on the selected dimension instead of falling back to selector-order inference.
+- translator/metrics: make `rate` and `bytes_rate` preserve Loki per-second semantics via window normalization, make parser+unwrap metric paths preserve Loki-like cardinality with outer-aggregation composition, emit VictoriaLogs-compatible byte aggregations via `sum_len(_msg)`, and implement `stdvar_over_time` via proxy-side `stddev^2` composition to avoid backend `stdvar` parser failures.
+- proxy/binary-metrics: fix scalar and binary post-processing to mutate both legacy `results` payloads and Prometheus-style `data.result` payloads (including instant-vector `value` samples), preventing silent no-op arithmetic on valid `stats_query` responses.
+
+### Tests
+
+- drilldown/volume: add regression coverage for inferred non-service target labels (no synthetic `unknown_service`) and for Drilldown `fieldBy` fallback mapping on both vector and matrix volume endpoints.
+- compat/matrix: add operation/filter/function matrix coverage across translator and proxy scalar paths, with deterministic e2e checks for binary scalar operators, filter operators, and metric function families (cross-engine parity for compatible functions plus proxy-local expected-value checks where semantics intentionally differ), and add scalar/binary fuzz coverage for response-shape robustness.
+
 ## [1.11.0] - 2026-04-21
 
 ### Bug Fixes

docs/translation-reference.md

@@ -73,29 +73,29 @@ These stages are executed at the proxy level (VL has no native equivalents):
 
 | LogQL | LogsQL |
 |---|---|
-| `rate({...}[5m])` | `... \| stats rate()` |
+| `rate({...}[5m])` | `stats count()` + `math` normalization by window seconds, then `stats sum(...)` per grouping |
 | `count_over_time({...}[5m])` | `... \| stats count()` |
-| `bytes_over_time({...}[5m])` | `... \| stats sum(len(_msg))` |
-| `bytes_rate({...}[5m])` | `... \| stats rate_sum(len(_msg))` |
+| `bytes_over_time({...}[5m])` | `... \| stats sum_len(_msg)` |
+| `bytes_rate({...}[5m])` | `stats sum_len(_msg)` + `math` normalization by window seconds, then `stats sum(...)` per grouping |
 | `sum_over_time({...} \| unwrap f [5m])` | `... \| stats sum(f)` |
 | `avg_over_time({...} \| unwrap f [5m])` | `... \| stats avg(f)` |
 | `max_over_time({...} \| unwrap f [5m])` | `... \| stats max(f)` |
 | `min_over_time({...} \| unwrap f [5m])` | `... \| stats min(f)` |
 | `first_over_time({...} \| unwrap f [5m])` | `... \| stats first(f)` |
 | `last_over_time({...} \| unwrap f [5m])` | `... \| stats last(f)` |
 | `stddev_over_time({...} \| unwrap f [5m])` | `... \| stats stddev(f)` |
-| `stdvar_over_time({...} \| unwrap f [5m])` | `... \| stats stdvar(f)` |
+| `stdvar_over_time({...} \| unwrap f [5m])` | proxy binary expression: `(... \| stats stddev(f)) ^ 2` |
 | `quantile_over_time(0.95, {...} \| unwrap f [5m])` | `... \| stats quantile(0.95, f)` |
 | `absent_over_time({...}[5m])` | `... \| stats count()` |
 
 ### Outer Aggregations
 
 | LogQL | LogsQL |
 |---|---|
-| `sum(rate({...}[5m]))` | `... \| stats rate()` |
-| `sum(rate({...}[5m])) by (x)` | `... \| stats by (x) rate()` |
-| `avg(rate({...}[5m])) by (x)` | `... \| stats by (x) rate()` |
-| `topk(10, rate({...}[5m]))` | `... \| stats rate()` |
+| `sum(rate({...}[5m]))` | normalized per-stream/per-group rate, then proxy applies outer aggregation where supported |
+| `sum(rate({...}[5m])) by (x)` | `... \| stats by (x) count() as __lvp_inner \| math __lvp_inner/window as __lvp_rate \| stats by (x) sum(__lvp_rate)` |
+| `avg(rate({...}[5m])) by (x)` | same normalized-rate path grouped by `(x)` |
+| `topk(10, rate({...}[5m]))` | normalized-rate path with stream grouping; top-level selection remains simplified |
 
 Supported: `sum`, `avg`, `max`, `min`, `count`, `topk`, `bottomk`, `stddev`, `stdvar`, `sort`, `sort_desc`.
 

internal/proxy/drilldown_compat_test.go

@@ -407,6 +407,12 @@ func TestDrilldown_IndexVolume_InfersPrimaryTargetLabelForAdditionalTabs(t *test
 	if len(result) != 2 {
 		t.Fatalf("expected cluster buckets, got %v", result)
 	}
+	for _, item := range result {
+		metric := item.(map[string]interface{})["metric"].(map[string]interface{})
+		if _, ok := metric["service_name"]; ok {
+			t.Fatalf("expected inferred cluster volume metrics to omit synthetic unknown service_name, got %v", metric)
+		}
+	}
 }
 
 func TestDrilldown_IndexVolume_TranslatesInferredTargetLabelMetrics(t *testing.T) {
@@ -461,6 +467,140 @@ func TestDrilldown_IndexVolume_TranslatesInferredTargetLabelMetrics(t *testing.T
 	}
 }
 
+func TestDrilldown_IndexVolume_UsesDrilldownFieldByFallbackForTargetLabels(t *testing.T) {
+	var receivedField string
+	vlBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/select/logsql/hits" {
+			t.Fatalf("unexpected backend path %s", r.URL.Path)
+		}
+		receivedField = r.URL.Query().Get("field")
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"hints": map[string]interface{}{},
+			"hits": []map[string]interface{}{
+				{
+					"fields":     map[string]string{"method": "GET"},
+					"timestamps": []string{"2026-04-04T17:18:49Z"},
+					"values":     []int{4},
+				},
+			},
+		})
+	}))
+	defer vlBackend.Close()
+
+	p := newGapTestProxy(t, vlBackend.URL)
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest(
+		"GET",
+		"/loki/api/v1/index/volume?query=%7Bcluster%3D~%60.%2B%60%7D&start=1&end=2&var-fieldBy=method&drillDownLabel=method",
+		nil,
+	)
+	p.handleVolume(w, r)
+
+	if receivedField != "method" {
+		t.Fatalf("expected drilldown fieldBy fallback to drive target label mapping, got %q", receivedField)
+	}
+
+	var resp map[string]interface{}
+	mustUnmarshal(t, w.Body.Bytes(), &resp)
+	data := assertDataIsObject(t, resp)
+	result := assertResultIsArray(t, data)
+	if len(result) != 1 {
+		t.Fatalf("expected one method bucket, got %v", result)
+	}
+	metric := result[0].(map[string]interface{})["metric"].(map[string]interface{})
+	if metric["method"] != "GET" {
+		t.Fatalf("expected method grouping bucket in metric, got %v", metric)
+	}
+}
+
+func TestDrilldown_IndexVolumeRange_InfersPrimaryTargetLabelWithoutUnknownService(t *testing.T) {
+	var receivedField string
+	vlBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/select/logsql/hits" {
+			t.Fatalf("unexpected backend path %s", r.URL.Path)
+		}
+		receivedField = r.URL.Query().Get("field")
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"hints": map[string]interface{}{},
+			"hits": []map[string]interface{}{
+				{
+					"fields":     map[string]string{"cluster": "us-east-1"},
+					"timestamps": []string{"2026-04-04T17:18:49Z"},
+					"values":     []int{12},
+				},
+			},
+		})
+	}))
+	defer vlBackend.Close()
+
+	p := newGapTestProxy(t, vlBackend.URL)
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("GET", "/loki/api/v1/index/volume_range?query=%7Bcluster%3D~%60.%2B%60%7D&start=1&end=2&step=60", nil)
+	p.handleVolumeRange(w, r)
+
+	if receivedField != "cluster" {
+		t.Fatalf("expected inferred volume_range field=cluster, got %q", receivedField)
+	}
+
+	var resp map[string]interface{}
+	mustUnmarshal(t, w.Body.Bytes(), &resp)
+	data := assertDataIsObject(t, resp)
+	result := assertResultIsArray(t, data)
+	if len(result) != 1 {
+		t.Fatalf("expected one cluster series, got %v", result)
+	}
+	metric := result[0].(map[string]interface{})["metric"].(map[string]interface{})
+	if _, ok := metric["service_name"]; ok {
+		t.Fatalf("expected inferred cluster volume_range metric to omit synthetic unknown service_name, got %v", metric)
+	}
+}
+
+func TestDrilldown_IndexVolumeRange_UsesDrilldownFieldByFallbackForTargetLabels(t *testing.T) {
+	var receivedField string
+	vlBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/select/logsql/hits" {
+			t.Fatalf("unexpected backend path %s", r.URL.Path)
+		}
+		receivedField = r.URL.Query().Get("field")
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"hints": map[string]interface{}{},
+			"hits": []map[string]interface{}{
+				{
+					"fields":     map[string]string{"method": "POST"},
+					"timestamps": []string{"2026-04-04T17:18:49Z"},
+					"values":     []int{6},
+				},
+			},
+		})
+	}))
+	defer vlBackend.Close()
+
+	p := newGapTestProxy(t, vlBackend.URL)
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest(
+		"GET",
+		"/loki/api/v1/index/volume_range?query=%7Bcluster%3D~%60.%2B%60%7D&start=1&end=2&step=60&var-fieldBy=method&drillDownLabel=method",
+		nil,
+	)
+	p.handleVolumeRange(w, r)
+
+	if receivedField != "method" {
+		t.Fatalf("expected drilldown fieldBy fallback to drive volume_range target label mapping, got %q", receivedField)
+	}
+
+	var resp map[string]interface{}
+	mustUnmarshal(t, w.Body.Bytes(), &resp)
+	data := assertDataIsObject(t, resp)
+	result := assertResultIsArray(t, data)
+	if len(result) != 1 {
+		t.Fatalf("expected one method series, got %v", result)
+	}
+	metric := result[0].(map[string]interface{})["metric"].(map[string]interface{})
+	if metric["method"] != "POST" {
+		t.Fatalf("expected method grouping series in volume_range metric, got %v", metric)
+	}
+}
+
 func TestDrilldown_IndexVolumeRange_TargetLabelsDetectedLevelUsesDerivedAggregation(t *testing.T) {
 	vlBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != "/select/logsql/hits" {

internal/proxy/fuzz_test.go

@@ -2,6 +2,8 @@ package proxy
 
 import (
 	"encoding/json"
+	"fmt"
+	"strconv"
 	"testing"
 )
 
@@ -148,3 +150,52 @@ func FuzzNormalizeMetadataPairs(f *testing.F) {
 		}
 	})
 }
+
+// FuzzCombineBinaryMetricResultsScalarPath validates scalar/vector merge shaping for all operators.
+func FuzzCombineBinaryMetricResultsScalarPath(f *testing.F) {
+	ops := []string{"+", "-", "*", "/", "%", "^", "==", "!=", ">", "<", ">=", "<=", "unknown"}
+	for _, op := range ops {
+		f.Add(op, 10.0, 3.0, false, true)
+		f.Add(op, 10.0, 3.0, true, false)
+		f.Add(op, 10.0, 3.0, false, false)
+	}
+
+	f.Fuzz(func(t *testing.T, op string, metricValue, scalarValue float64, leftScalar, rightScalar bool) {
+		if !isFinite(metricValue) || !isFinite(scalarValue) {
+			return
+		}
+
+		metricBody := []byte(fmt.Sprintf(
+			`{"results":[{"metric":{"app":"fuzz"},"values":[[1,"%s"]]}]}`,
+			strconv.FormatFloat(metricValue, 'f', -1, 64),
+		))
+		scalarBody := []byte(fmt.Sprintf(
+			`{"status":"success","data":{"resultType":"scalar","result":[0,"%s"]}}`,
+			strconv.FormatFloat(scalarValue, 'f', -1, 64),
+		))
+
+		leftBody := metricBody
+		rightBody := metricBody
+		leftQL := ""
+		rightQL := ""
+
+		if leftScalar {
+			leftBody = scalarBody
+			leftQL = strconv.FormatFloat(scalarValue, 'f', -1, 64)
+		}
+		if rightScalar {
+			rightBody = scalarBody
+			rightQL = strconv.FormatFloat(scalarValue, 'f', -1, 64)
+		}
+
+		out := combineBinaryMetricResults(leftBody, rightBody, op, "matrix", leftScalar, rightScalar, leftQL, rightQL)
+		var payload map[string]interface{}
+		if err := json.Unmarshal(out, &payload); err != nil {
+			t.Fatalf("combined output must stay valid JSON: %v; body=%s", err, string(out))
+		}
+		status, _ := payload["status"].(string)
+		if status == "" {
+			t.Fatalf("combined output missing status field: %#v", payload)
+		}
+	})
+}