From 405c463a633b0af90bfd37f55c82c568e17b2e36 Mon Sep 17 00:00:00 2001
From: szibis <szibis@users.noreply.github.com>
Date: Mon, 12 Oct 2026 12:59:23 +0200
Subject: [PATCH 1/4] fix: a vector join fails with Loki's duplicate-series
 text at the step where Loki fails

The right-hand duplicate error joined its two halves with "; " where Loki
v3.7.7 uses ";" (evaluator.go vectorBinop). The cardinality checks ran over
every step before the group-label checks, where Loki stops at the first
failing step. Adds a differential test against a reference written from
vectorBinop, an ignoring() edge table, handler tests on range and instant
queries and the live comparison with Loki.
---
 CHANGELOG.md                                  |  29 ++
 .../semantics/binary-ignoring-matching.yaml   |  28 +-
 .../proxy/binary_ignoring_matching_test.go    | 332 ++++++++++++++++
 internal/proxy/binary_logql.go                |   8 +-
 internal/proxy/binary_matching.go             |  11 +-
 .../binary_matching_differential_test.go      | 364 ++++++++++++++++++
 internal/proxy/vector_matching.go             | 181 +++++----
 .../binary_ignoring_matching_test.go          | 226 +++++++++++
 8 files changed, 1093 insertions(+), 86 deletions(-)
 create mode 100644 internal/proxy/binary_ignoring_matching_test.go
 create mode 100644 internal/proxy/binary_matching_differential_test.go
 create mode 100644 test/e2e-compat/binary_ignoring_matching_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index feca92cbd..ee0dfba39 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **A vector-to-vector join fails with Loki's exact error text, at the step where
+  Loki fails.** When the right operand of `A + ignoring(machine) B` held several
+  series for one match key the proxy answered `found duplicate series on the
+  right hand-side; many-to-many matching not allowed: ...`, with a space after
+  the semicolon; Loki v3.7.7 joins the two halves with `;` alone
+  (pkg/logql/evaluator.go vectorBinop, :974-975), so a client matching the
+  message saw a different string (the `multiple matches for labels` texts already
+  matched). Loki joins one evaluation at a time and stops at the first whose
+  join fails (pkg/logql/engine.go reads the evaluator's error after every step);
+  the proxy checked the right-hand duplicates of every step before the left-hand
+  ones and before the group-label collisions, so a range query that fails at
+  several steps could name another error than Loki. The join now raises the error
+  of the earliest failing step, right-hand duplicates before left-hand multiple
+  matches inside a step. Pinned by a differential test over 6,000 generated
+  operand pairs and modifier sets (none, `on`, `ignoring`, `group_left`,
+  `group_right`, with and without `bool`) against a reference written from
+  vectorBinop, matchingSignature, resultMetric and MergeBinOp, an edge table for
+  `ignoring` over a label absent on both sides, on one side and on the right
+  only, and handler tests on range and instant queries; the live comparison with
+  Loki is `TestCompat_BinaryIgnoringAbsentLabel` (#TBD). Not changed: the query
+  from the conformance row `semantics/binary-ignoring-matching`, `sum by
+  (cluster) (sum by (cluster) (rate(..)) + ignoring(machine) sum by (cluster,
+  machine) (rate(..)))`, still answers the first operand's raw count (about 150
+  times Loki's value) because an outer aggregation over a binary expression is
+  translated from its first operand; the join itself matches Loki, and the row
+  stays open for that path.
+
 ## [2.11.2] - 2026-10-12
 
 ### CI
diff --git a/conformance/registry/cases/semantics/binary-ignoring-matching.yaml b/conformance/registry/cases/semantics/binary-ignoring-matching.yaml
index 89c210db1..424106008 100644
--- a/conformance/registry/cases/semantics/binary-ignoring-matching.yaml
+++ b/conformance/registry/cases/semantics/binary-ignoring-matching.yaml
@@ -8,6 +8,26 @@ why: |
   ignoring(machine) sum by (cluster, machine) (rate({namespace="prod"}[5m])))`
   is one series per cluster on Loki (2) with the sum of both sides; the proxy
   returns 1 series with 150x Loki's value.
+
+  Root cause (measured against the real handler with a VictoriaLogs stand-in):
+  the join is not the defect. `sum by (cluster) (rate(..)) + ignoring(machine)
+  sum by (cluster, machine) (rate(..))` on its own is evaluated by the binary
+  evaluator and gives Loki's series: the ignored label is dropped from the
+  match key and from the result, an empty value counts as absent (Loki
+  pkg/logql/evaluator.go matchingSignature :945, resultMetric :1096). Under the
+  outer `sum by (cluster) (...)` the query never reaches the evaluator: the
+  translator takes the first operand of the binary and sends only
+  `stats by (cluster) count()` to VictoriaLogs, so the answer is that operand's
+  raw count, with no rate division and no second operand (a count of C against
+  Loki's 2C/300 is the 150x). The same happens for any outer aggregation over a
+  binary (`sum(A / B)`), with or without a vector-matching modifier.
+
+  Fixed in the matching code, found while pinning it to Loki: the duplicate
+  series text had a space after the semicolon (Loki joins the two halves of the
+  text with ";" alone, evaluator.go:974-975); and the join checked every
+  step's right-hand duplicates before any left-hand ones and before the
+  group-label collisions, where Loki stops at the first step whose join fails
+  (engine.go reads the evaluator's error after every step).
 request:
   method: GET
   path: /loki/api/v1/query_range
@@ -28,10 +48,10 @@ gap:
   status: open
   impact: explore-visible
   effort: S
-  area: 'internal/proxy/vector_matching.go (ignoring with a label absent on both sides)'
-  loki: 'pkg/logql/evaluator.go binary vector matching (ignoring)'
-  proxy: 'wrong join'
-  planned_test: 'TestCompat_BinaryIgnoringAbsentLabel (test/e2e-compat, group semantics)'
+  area: 'translator and handleQueryRange/handleQuery: an outer aggregation over a binary expression (the binary evaluator itself matches Loki)'
+  loki: 'pkg/logql/evaluator.go:791-1128 newBinOpStepEvaluator, matchingSignature, vectorBinop, resultMetric; pkg/logql/engine.go step loop'
+  proxy: 'sum by (..) (A + ignoring(..) B) is translated from its first operand alone: a raw count, no join'
+  planned_test: 'TestCompat_BinaryIgnoringOuterSum (the row's shape, test/e2e-compat, group semantics) and TestCompat_BinaryIgnoringAbsentLabel (the join itself); TestBinaryMatchingDifferential, TestBinaryMatchingErrorTextAndStepOrder, TestBinaryIgnoringMatchingEdgeTable, TestBinaryIgnoringThroughHandler (internal/proxy)'
   clusters:
     - 354c7d83
     - 8fef8a78
diff --git a/internal/proxy/binary_ignoring_matching_test.go b/internal/proxy/binary_ignoring_matching_test.go
new file mode 100644
index 000000000..b326e98d0
--- /dev/null
+++ b/internal/proxy/binary_ignoring_matching_test.go
@@ -0,0 +1,332 @@
+package proxy
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/ReliablyObserve/Loki-VL-proxy/internal/translator"
+)
+
+// The vector matching errors, byte for byte as Loki v3.7.7 raises them in
+// pkg/logql/evaluator.go vectorBinop (:974-975 joins the two halves of the
+// duplicate-series text with ";" and no space; :998 and :1007).
+const (
+	lokiDuplicateRight = "found duplicate series on the right hand-side;many-to-many matching not allowed: matching labels must be unique on one side"
+	lokiDuplicateLeft  = "found duplicate series on the left hand-side;many-to-many matching not allowed: matching labels must be unique on one side"
+	lokiManyToOne      = "multiple matches for labels: many-to-one matching must be explicit (group_left/group_right)"
+	lokiGroupingUnique = "multiple matches for labels: grouping labels must ensure unique matches"
+)
+
+type matchSeries struct {
+	labels map[string]string
+	points [][]any
+}
+
+func matchBody(series ...matchSeries) []byte {
+	labels := make([]map[string]string, len(series))
+	points := make([][][]any, len(series))
+	for i, s := range series {
+		labels[i], points[i] = s.labels, s.points
+	}
+	return binaryTestBody(labels, points...)
+}
+
+func at(ts int, value string) [][]any { return [][]any{{ts, value}} }
+
+func pts(pairs ...any) [][]any {
+	out := make([][]any, 0, len(pairs)/2)
+	for i := 0; i+1 < len(pairs); i += 2 {
+		out = append(out, []any{pairs[i], pairs[i+1]})
+	}
+	return out
+}
+
+func decodeMatchResult(t *testing.T, body []byte) map[string][][]any {
+	t.Helper()
+	var response struct {
+		Data struct {
+			Result []struct {
+				Metric map[string]string
+				Values [][]any
+			}
+		}
+	}
+	if err := json.Unmarshal(body, &response); err != nil {
+		t.Fatal(err)
+	}
+	got := map[string][][]any{}
+	for _, series := range response.Data.Result {
+		got[binaryLabelKey(series.Metric)] = series.Values
+	}
+	return got
+}
+
+// Loki evaluates one step at a time and stops at the first step whose join
+// fails (pkg/logql/engine.go: Error() is read after every Next()). Within a
+// step vectorBinop rejects a duplicate on the right hand-side before it
+// looks at the left, and names the side that is the "one" side of the
+// grouping.
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range
+func TestBinaryMatchingErrorTextAndStepOrder(t *testing.T) {
+	s := func(labels map[string]string, points [][]any) matchSeries { return matchSeries{labels, points} }
+	a, b := map[string]string{"app": "a", "pod": "1"}, map[string]string{"app": "a", "pod": "2"}
+	one := map[string]string{"app": "a"}
+	ignoringPod := &translator.VectorMatchInfo{Ignoring: []string{"pod"}}
+	onApp := &translator.VectorMatchInfo{MatchOn: true, On: []string{"app"}}
+	for _, tc := range []struct {
+		name        string
+		left, right []byte
+		vm          *translator.VectorMatchInfo
+		want        string
+	}{
+		{"duplicate_right_one_step", matchBody(s(one, at(1, "1"))), matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), ignoringPod, lokiDuplicateRight},
+		{"duplicate_left_is_many_to_one", matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), matchBody(s(one, at(1, "1"))), ignoringPod, lokiManyToOne},
+		{"duplicate_both_sides_same_step_names_right", matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), onApp, lokiDuplicateRight},
+		{"earlier_step_many_to_one_wins_over_later_right_duplicate",
+			matchBody(s(a, at(1, "1")), s(b, at(1, "2")), s(one, at(2, "1"))),
+			matchBody(s(one, at(1, "1")), s(a, at(2, "1")), s(b, at(2, "2"))), ignoringPod, lokiManyToOne},
+		{"earlier_step_right_duplicate_wins_over_later_many_to_one",
+			matchBody(s(one, at(1, "1")), s(a, at(2, "1")), s(b, at(2, "2"))),
+			matchBody(s(a, at(1, "1")), s(b, at(1, "2")), s(one, at(2, "1"))), ignoringPod, lokiDuplicateRight},
+		{"unmatched_left_duplicates_are_not_an_error_before_a_later_right_duplicate",
+			matchBody(s(map[string]string{"app": "x", "pod": "1"}, at(1, "1")), s(map[string]string{"app": "x", "pod": "2"}, at(1, "2"))),
+			matchBody(s(a, at(1, "1")), s(a, at(2, "1"))), ignoringPod, ""},
+		{"group_left_duplicate_right_is_the_one_side",
+			matchBody(s(a, at(1, "1"))), matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), &translator.VectorMatchInfo{MatchOn: true, On: []string{"app"}, GroupSide: "group_left"}, lokiDuplicateRight},
+		{"group_right_duplicate_left_is_the_one_side",
+			matchBody(s(a, at(1, "1")), s(b, at(1, "2"))), matchBody(s(a, at(1, "1"))), &translator.VectorMatchInfo{MatchOn: true, On: []string{"app"}, GroupSide: "group_right"}, lokiDuplicateLeft},
+		{"group_left_colliding_include",
+			matchBody(s(map[string]string{"app": "a", "zone": "one"}, at(1, "1")), s(map[string]string{"app": "a", "zone": "two"}, at(1, "2"))),
+			matchBody(s(map[string]string{"app": "a", "zone": "same"}, at(1, "1"))),
+			&translator.VectorMatchInfo{MatchOn: true, On: []string{"app"}, GroupSide: "group_left", GroupLeft: []string{"zone"}}, lokiGroupingUnique},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := matchBinaryMetricResults(tc.left, tc.right, "+", "matrix", tc.vm, false)
+			switch {
+			case tc.want == "" && err != nil:
+				t.Fatalf("unexpected error %q", err)
+			case tc.want != "" && (err == nil || err.Error() != tc.want):
+				t.Fatalf("error = %v, want %q", err, tc.want)
+			}
+		})
+	}
+}
+
+// ignoring(label) over operands where the label is absent, present on one
+// side, or present on both: the join key and the labels of the result follow
+// Loki's matchingSignature and resultMetric (evaluator.go:945-953, :1096-1128).
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range
+func TestBinaryIgnoringMatchingEdgeTable(t *testing.T) {
+	s := func(labels map[string]string, points [][]any) matchSeries { return matchSeries{labels, points} }
+	c1, c2 := map[string]string{"cluster": "c1"}, map[string]string{"cluster": "c2"}
+	c1m1, c1m2 := map[string]string{"cluster": "c1", "machine": "m1"}, map[string]string{"cluster": "c1", "machine": "m2"}
+	ignoring := func(labels ...string) *translator.VectorMatchInfo {
+		return &translator.VectorMatchInfo{Ignoring: labels}
+	}
+	for _, tc := range []struct {
+		name        string
+		left, right []byte
+		op          string
+		vm          *translator.VectorMatchInfo
+		boolean     bool
+		want        map[string][][]any
+	}{
+		{"label_absent_on_both_sides",
+			matchBody(s(c1, pts(1, "3", 2, "4")), s(c2, at(1, "5"))), matchBody(s(c1, pts(1, "30", 2, "40")), s(c2, at(1, "50"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33", float64(2), "44"), `{"cluster":"c2"}`: pts(float64(1), "55")}},
+		{"label_empty_on_the_right_counts_as_absent",
+			matchBody(s(c1, at(1, "3"))), matchBody(s(map[string]string{"cluster": "c1", "machine": ""}, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"label_on_the_right_only_is_dropped_from_the_key",
+			matchBody(s(c1, at(1, "3"))), matchBody(s(c1m1, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"label_on_the_left_only_is_dropped_from_the_result",
+			matchBody(s(c1m1, at(1, "3"))), matchBody(s(c1, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"different_values_of_the_ignored_label_still_join",
+			matchBody(s(c1m1, at(1, "3"))), matchBody(s(c1m2, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"unmatched_series_are_dropped",
+			matchBody(s(c1, at(1, "3")), s(c2, at(1, "4"))), matchBody(s(c1, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"ignoring_a_label_both_sides_carry_with_the_same_value",
+			matchBody(s(c1m1, at(1, "3"))), matchBody(s(c1m1, at(1, "30"))), "+", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"ignoring_nothing_needs_equal_labels",
+			matchBody(s(c1, at(1, "3"))), matchBody(s(c1m1, at(1, "30"))), "+", ignoring(), false,
+			map[string][][]any{}},
+		{"comparison_filter_keeps_the_left_value_and_drops_the_ignored_label",
+			matchBody(s(c1m1, pts(1, "5", 2, "1"))), matchBody(s(c1, pts(1, "3", 2, "3"))), ">", ignoring("machine"), false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "5")}},
+		{"comparison_bool_returns_zero_and_one_and_drops_the_ignored_label",
+			matchBody(s(c1m1, pts(1, "5", 2, "1"))), matchBody(s(c1, pts(1, "3", 2, "3"))), ">", ignoring("machine"), true,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "1", float64(2), "0")}},
+		{"on_keeps_only_the_on_labels_of_the_left",
+			matchBody(s(c1m1, at(1, "3"))), matchBody(s(c1, at(1, "30"))), "+", &translator.VectorMatchInfo{MatchOn: true, On: []string{"cluster"}}, false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+		{"group_left_keeps_every_left_label_and_adds_the_included_one",
+			matchBody(s(c1m1, at(1, "3")), s(c1m2, at(1, "4"))), matchBody(s(map[string]string{"cluster": "c1", "dc": "x"}, at(1, "30"))), "+",
+			&translator.VectorMatchInfo{Ignoring: []string{"machine", "dc"}, GroupSide: "group_left", GroupLeft: []string{"dc"}}, false,
+			map[string][][]any{`{"cluster":"c1","dc":"x","machine":"m1"}`: pts(float64(1), "33"), `{"cluster":"c1","dc":"x","machine":"m2"}`: pts(float64(1), "34")}},
+		{"group_left_include_missing_on_the_one_side_removes_the_label",
+			matchBody(s(map[string]string{"cluster": "c1", "dc": "y"}, at(1, "3"))), matchBody(s(c1, at(1, "30"))), "+",
+			&translator.VectorMatchInfo{Ignoring: []string{"dc"}, GroupSide: "group_left", GroupLeft: []string{"dc"}}, false,
+			map[string][][]any{`{"cluster":"c1"}`: pts(float64(1), "33")}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			body, err := matchBinaryMetricResults(tc.left, tc.right, tc.op, "matrix", tc.vm, tc.boolean)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got := decodeMatchResult(t, body); !reflect.DeepEqual(got, tc.want) {
+				t.Fatalf("got %s, want %v", body, tc.want)
+			}
+		})
+	}
+}
+
+// vlOperandServer answers each operand of a binary expression with the
+// VictoriaLogs stats_query_range body registered for its `stats by (...)`
+// clause. Counts are per five minute window, so a rate is the count / 300.
+func vlOperandServer(t *testing.T, byClause map[string]string) *httptest.Server {
+	t.Helper()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = r.ParseForm()
+		query := r.Form.Get("query")
+		for clause, body := range byClause {
+			if strings.Contains(query, clause) {
+				w.Header().Set("Content-Type", "application/json")
+				_, _ = w.Write([]byte(body))
+				return
+			}
+		}
+		t.Errorf("unexpected VictoriaLogs query %q", query)
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	t.Cleanup(server.Close)
+	return server
+}
+
+func statsMatrix(series ...string) string {
+	return `{"status":"success","data":{"resultType":"matrix","result":[` + strings.Join(series, ",") + `]}}`
+}
+
+func statsSeries(labels string, values ...string) string {
+	points := make([]string, len(values))
+	for i, v := range values {
+		points[i] = `[` + []string{"1700000100", "1700000160"}[i] + `,"` + v + `"]`
+	}
+	return `{"metric":{"__name__":"c",` + labels + `},"values":[` + strings.Join(points, ",") + `]}`
+}
+
+// The operands of the registry row, `sum by (cluster) (rate(..)) + ignoring
+// (machine) sum by (cluster, machine) (rate(..))`, over a stream set that has
+// no `machine` label: one series per cluster holding the sum of both operands,
+// each read from its own VictoriaLogs query. With a `machine` label the right
+// operand holds several series per cluster, and Loki rejects the join with its
+// duplicate-series text.
+//
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range, loki_api_v1_query
+func TestBinaryIgnoringThroughHandler(t *testing.T) {
+	const (
+		left   = `sum by (cluster) (rate({namespace="prod"}[5m]))`
+		right  = `sum by (cluster, machine) (rate({namespace="prod"}[5m]))`
+		joined = left + ` + ignoring(machine) ` + right
+	)
+	// The operands differ in value so that reading one of them twice fails.
+	absentRange := map[string]string{
+		"stats by (cluster, machine)": statsMatrix(statsSeries(`"cluster":"c1"`, "1500", "3000"), statsSeries(`"cluster":"c2"`, "750", "900")),
+		"stats by (cluster)":          statsMatrix(statsSeries(`"cluster":"c1"`, "300", "600"), statsSeries(`"cluster":"c2"`, "150", "150")),
+	}
+	presentRange := map[string]string{
+		"stats by (cluster, machine)": statsMatrix(statsSeries(`"cluster":"c1","machine":"m1"`, "300", "300"), statsSeries(`"cluster":"c1","machine":"m2"`, "300", "300")),
+		"stats by (cluster)":          statsMatrix(statsSeries(`"cluster":"c1"`, "300", "600")),
+	}
+	vector := func(series ...string) string {
+		return `{"status":"success","data":{"resultType":"vector","result":[` + strings.Join(series, ",") + `]}}`
+	}
+	sample := func(labels, value string) string {
+		return `{"metric":{"__name__":"c",` + labels + `},"value":[1700000160,"` + value + `"]}`
+	}
+	absentInstant := map[string]string{
+		"stats by (cluster, machine)": vector(sample(`"cluster":"c1"`, "1500"), sample(`"cluster":"c2"`, "750")),
+		"stats by (cluster)":          vector(sample(`"cluster":"c1"`, "300"), sample(`"cluster":"c2"`, "150")),
+	}
+	presentInstant := map[string]string{
+		"stats by (cluster, machine)": vector(sample(`"cluster":"c1","machine":"m1"`, "300"), sample(`"cluster":"c1","machine":"m2"`, "300")),
+		"stats by (cluster)":          vector(sample(`"cluster":"c1"`, "300")),
+	}
+	for _, instant := range []bool{false, true} {
+		absent, present, name := absentRange, presentRange, "range"
+		if instant {
+			absent, present, name = absentInstant, presentInstant, "instant"
+		}
+		run := func(t *testing.T, vl map[string]string, query string) *httptest.ResponseRecorder {
+			p := newTestProxy(t, vlOperandServer(t, vl).URL)
+			mux := http.NewServeMux()
+			p.RegisterRoutes(mux)
+			endpoint, params := "query_range", url.Values{"query": {query}, "start": {"1700000100"}, "end": {"1700000160"}, "step": {"60"}}
+			if instant {
+				endpoint, params = "query", url.Values{"query": {query}, "time": {"1700000160"}}
+			}
+			r := httptest.NewRequest("GET", "/loki/api/v1/"+endpoint+"?"+params.Encode(), nil)
+			r.Header.Set("X-Scope-OrgID", "0")
+			r.Header.Set("X-Loki-Response-Encoding-Flags", "categorize-labels")
+			w := httptest.NewRecorder()
+			mux.ServeHTTP(w, r)
+			return w
+		}
+		points := func(t *testing.T, w *httptest.ResponseRecorder) map[string][][]any {
+			t.Helper()
+			var response struct {
+				Data struct {
+					Result []struct {
+						Metric map[string]string
+						Value  []any
+						Values [][]any
+					}
+				}
+			}
+			if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || w.Code != 200 {
+				t.Fatalf("%d %s", w.Code, w.Body)
+			}
+			got := map[string][][]any{}
+			for _, series := range response.Data.Result {
+				values := series.Values
+				if instant {
+					values = [][]any{series.Value}
+				}
+				got[binaryLabelKey(series.Metric)] = values
+			}
+			return got
+		}
+		t.Run(name+"/machine_absent", func(t *testing.T) {
+			a, b, both := points(t, run(t, absent, left)), points(t, run(t, absent, right)), points(t, run(t, absent, joined))
+			if len(a) != 2 || len(b) != 2 || len(both) != 2 {
+				t.Fatalf("want one series per cluster: %v %v %v", a, b, both)
+			}
+			for key, values := range a {
+				if len(values) == 0 || len(b[key]) != len(values) || len(both[key]) != len(values) {
+					t.Fatalf("series %s: left %v right %v join %v", key, values, b[key], both[key])
+				}
+				for i := range values {
+					l, r, j := parsePointValue(values[i][1]), parsePointValue(b[key][i][1]), parsePointValue(both[key][i][1])
+					if l == r || j != l+r {
+						t.Fatalf("series %s point %d: join = %v, want left %v + right %v", key, i, j, l, r)
+					}
+				}
+			}
+		})
+		t.Run(name+"/machine_present_on_the_right", func(t *testing.T) {
+			w := run(t, present, joined)
+			if w.Code != 500 || !strings.Contains(w.Body.String(), lokiDuplicateRight) {
+				t.Fatalf("status %d, want 500 with %q: %s", w.Code, lokiDuplicateRight, w.Body)
+			}
+		})
+	}
+}
diff --git a/internal/proxy/binary_logql.go b/internal/proxy/binary_logql.go
index f5f1578ae..5cfaf1fb7 100644
--- a/internal/proxy/binary_logql.go
+++ b/internal/proxy/binary_logql.go
@@ -332,18 +332,18 @@ func alignBinaryRangeRequest(r *http.Request) *http.Request {
 	return aligned
 }
 
-func validateBinaryVectorCardinality(ctx context.Context, left, right []byte, op string, vm *translator.VectorMatchInfo, leftScalar, rightScalar bool) error {
+func validateBinaryVectorCardinality(ctx context.Context, left, right []byte, op string, vm *translator.VectorMatchInfo, leftScalar, rightScalar bool) (float64, error) {
 	if leftScalar || rightScalar || op == "and" || op == "or" || op == "unless" {
-		return nil
+		return 0, nil
 	}
 	if vm == nil {
-		return validateVectorMatchCardinalityContext(ctx, left, right, nil, nil, false, false)
+		return firstVectorMatchFailure(ctx, left, right, nil, nil, false, false)
 	}
 	on := vm.On
 	if vm.MatchOn && on == nil {
 		on = []string{}
 	}
-	return validateVectorMatchCardinalityContext(ctx, left, right, on, vm.Ignoring,
+	return firstVectorMatchFailure(ctx, left, right, on, vm.Ignoring,
 		vm.GroupSide == "group_left" || len(vm.GroupLeft) > 0,
 		vm.GroupSide == "group_right" || len(vm.GroupRight) > 0)
 }
diff --git a/internal/proxy/binary_matching.go b/internal/proxy/binary_matching.go
index f5aa4eb19..fce8c64a8 100644
--- a/internal/proxy/binary_matching.go
+++ b/internal/proxy/binary_matching.go
@@ -170,8 +170,12 @@ func matchBinaryMetricResultsContext(ctx context.Context, left, right []byte, op
 			return nil, err
 		}
 	}
-	if err := validateBinaryVectorCardinality(ctx, left, right, op, vm, false, false); err != nil {
-		return nil, err
+	// Loki joins one evaluation at a time and stops at the first that fails:
+	// the cardinality failure is raised where it falls, not before an
+	// earlier evaluation's own error.
+	cardinalityTS, cardinalityErr := validateBinaryVectorCardinality(ctx, left, right, op, vm, false, false)
+	if _, rejected := cardinalityErr.(vectorMatchError); cardinalityErr != nil && !rejected {
+		return nil, cardinalityErr
 	}
 	lhs, err := binarySamplesByTime(ctx, left)
 	if err != nil {
@@ -198,6 +202,9 @@ func matchBinaryMetricResultsContext(ctx context.Context, left, right []byte, op
 		if err := ctx.Err(); err != nil {
 			return nil, err
 		}
+		if cardinalityErr != nil && ts == cardinalityTS {
+			return nil, cardinalityErr
+		}
 		appendSample := func(sample binarySample) error {
 			if err := checkBinaryOutputSample(ctx); err != nil {
 				return err
diff --git a/internal/proxy/binary_matching_differential_test.go b/internal/proxy/binary_matching_differential_test.go
new file mode 100644
index 000000000..f1cde906a
--- /dev/null
+++ b/internal/proxy/binary_matching_differential_test.go
@@ -0,0 +1,364 @@
+package proxy
+
+import (
+	"fmt"
+	"math/rand"
+	"reflect"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/ReliablyObserve/Loki-VL-proxy/internal/translator"
+)
+
+// This file pins the vector-to-vector join of the binary evaluator to Loki
+// v3.7.7 over generated operands and generated modifiers. The reference below
+// is written apart from the proxy, step by step as Loki evaluates it:
+// pkg/logql/engine.go reads the evaluator's error after every step and stops
+// at the first one; pkg/logql/evaluator.go BinOpStepEvaluator.Next hashes each
+// sample with matchingSignature (:945, on keeps the listed labels, ignoring
+// removes them, an empty label value is no label) and calls vectorBinop (:955):
+// the right vector (the left one after a group_right swap) must be unique by
+// signature or the join fails with "found duplicate series on the <side>
+// hand-side", a left sample that meets an already matched signature without a
+// group modifier fails with "multiple matches for labels: many-to-one ...",
+// and resultMetric (:1096) builds the labels of a sample; syntax.MergeBinOp
+// (pkg/logql/syntax/ast.go:1944) gives the value (a filtering comparison
+// keeps the left value, a bool comparison is 1 or 0).
+
+type refSample struct {
+	labels map[string]string
+	value  float64
+}
+
+type refMatching struct {
+	on      bool
+	labels  []string
+	card    string // "", "group_left", "group_right"
+	include []string
+}
+
+func refSignature(labels map[string]string, m refMatching) string {
+	keep := map[string]string{}
+	for k, v := range labels {
+		if v == "" {
+			continue
+		}
+		keep[k] = v
+	}
+	if m.on {
+		only := map[string]string{}
+		for _, l := range m.labels {
+			if v, ok := keep[l]; ok {
+				only[l] = v
+			}
+		}
+		keep = only
+	} else {
+		for _, l := range m.labels {
+			delete(keep, l)
+		}
+	}
+	return refLabelsKey(keep)
+}
+
+func refLabelsKey(labels map[string]string) string {
+	parts := make([]string, 0, len(labels))
+	for k, v := range labels {
+		if v != "" {
+			parts = append(parts, k+"\xff"+v)
+		}
+	}
+	sort.Strings(parts)
+	return strings.Join(parts, "\xfe")
+}
+
+func refResultLabels(many, one map[string]string, m refMatching) map[string]string {
+	out := map[string]string{}
+	for k, v := range many {
+		if v != "" {
+			out[k] = v
+		}
+	}
+	if m.card == "" {
+		if m.on {
+			for k := range out {
+				found := false
+				for _, l := range m.labels {
+					found = found || l == k
+				}
+				if !found {
+					delete(out, k)
+				}
+			}
+		} else {
+			for _, l := range m.labels {
+				delete(out, l)
+			}
+		}
+	}
+	for _, l := range m.include {
+		if v := one[l]; v != "" {
+			out[l] = v
+		} else {
+			delete(out, l)
+		}
+	}
+	return out
+}
+
+func refMerge(op string, l, r float64, returnBool bool) (float64, bool) {
+	switch op {
+	case "+":
+		return l + r, true
+	case "-":
+		return l - r, true
+	case "*":
+		return l * r, true
+	case "/":
+		return l / r, true // the generated divisors are never zero
+	}
+	var truth bool
+	switch op {
+	case "==":
+		truth = l == r
+	case "!=":
+		truth = l != r
+	case ">":
+		truth = l > r
+	case "<":
+		truth = l < r
+	case ">=":
+		truth = l >= r
+	case "<=":
+		truth = l <= r
+	}
+	if returnBool {
+		if truth {
+			return 1, true
+		}
+		return 0, true
+	}
+	return l, truth
+}
+
+// refStep is one evaluation of Loki's vectorBinop.
+func refStep(op string, m refMatching, returnBool bool, lhs, rhs []refSample) ([]refSample, error) {
+	oneToMany := m.card == "group_right"
+	if oneToMany {
+		lhs, rhs = rhs, lhs
+	}
+	right := map[string]refSample{}
+	for _, s := range rhs {
+		sig := refSignature(s.labels, m)
+		if _, dup := right[sig]; dup {
+			side := "right"
+			if oneToMany {
+				side = "left"
+			}
+			return nil, fmt.Errorf("found duplicate series on the %s hand-side;many-to-many matching not allowed: matching labels must be unique on one side", side)
+		}
+		right[sig] = s
+	}
+	matched := map[string]map[string]bool{}
+	var out []refSample
+	for _, s := range lhs {
+		sig := refSignature(s.labels, m)
+		r, found := right[sig]
+		if !found {
+			continue
+		}
+		metric := refResultLabels(s.labels, r.labels, m)
+		inserted, exists := matched[sig]
+		if m.card == "" {
+			if exists {
+				return nil, fmt.Errorf("multiple matches for labels: many-to-one matching must be explicit (group_left/group_right)")
+			}
+			matched[sig] = nil
+		} else {
+			key := refLabelsKey(metric)
+			if !exists {
+				inserted = map[string]bool{}
+				matched[sig] = inserted
+			} else if inserted[key] {
+				return nil, fmt.Errorf("multiple matches for labels: grouping labels must ensure unique matches")
+			}
+			inserted[key] = true
+		}
+		l, rr := s, r
+		if oneToMany {
+			l, rr = r, s
+		}
+		if v, ok := refMerge(op, l.value, rr.value, returnBool); ok {
+			out = append(out, refSample{metric, v})
+		}
+	}
+	return out, nil
+}
+
+// refMatrix evaluates every step in order, stops at the first failing one.
+func refMatrix(op string, m refMatching, returnBool bool, left, right map[float64][]refSample) (map[string][][]any, error) {
+	steps := map[float64]bool{}
+	for ts := range left {
+		steps[ts] = true
+	}
+	for ts := range right {
+		steps[ts] = true
+	}
+	axis := make([]float64, 0, len(steps))
+	for ts := range steps {
+		axis = append(axis, ts)
+	}
+	sort.Float64s(axis)
+	out := map[string][][]any{}
+	for _, ts := range axis {
+		samples, err := refStep(op, m, returnBool, left[ts], right[ts])
+		if err != nil {
+			return nil, err
+		}
+		for _, s := range samples {
+			key := binaryLabelKey(s.labels)
+			out[key] = append(out[key], []any{ts, strconv.FormatFloat(s.value, 'f', -1, 64)})
+		}
+	}
+	return out, nil
+}
+
+// refCollides reports two samples of one step that carry the same labels: Loki
+// then holds two points of one series at one time, which the generator avoids.
+func refCollides(samples []refSample) bool {
+	seen := map[string]bool{}
+	for _, s := range samples {
+		k := refLabelsKey(s.labels)
+		if seen[k] {
+			return true
+		}
+		seen[k] = true
+	}
+	return false
+}
+
+func genOperand(rng *rand.Rand, steps []float64) (map[float64][]refSample, []byte) {
+	apps, pods, machines := []string{"a", "b"}, []string{"", "p1", "p2"}, []string{"", "m1", "m2"}
+	type key struct{ app, pod, machine string }
+	all := []key{}
+	for _, a := range apps {
+		for _, p := range pods {
+			for _, m := range machines {
+				all = append(all, key{a, p, m})
+			}
+		}
+	}
+	rng.Shuffle(len(all), func(i, j int) { all[i], all[j] = all[j], all[i] })
+	n := rng.Intn(5)
+	byTime := map[float64][]refSample{}
+	var series []matchSeries
+	for _, k := range all[:n] {
+		labels := map[string]string{"app": k.app}
+		if k.pod != "" {
+			labels["pod"] = k.pod
+		}
+		if k.machine != "" {
+			labels["machine"] = k.machine
+		}
+		var points [][]any
+		for _, ts := range steps {
+			if rng.Intn(3) == 0 {
+				continue
+			}
+			v := float64(rng.Intn(9) + 1)
+			points = append(points, []any{ts, strconv.FormatFloat(v, 'f', -1, 64)})
+			byTime[ts] = append(byTime[ts], refSample{labels, v})
+		}
+		if len(points) > 0 {
+			series = append(series, matchSeries{labels, points})
+		}
+	}
+	return byTime, matchBody(series...)
+}
+
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range
+func TestBinaryMatchingDifferential(t *testing.T) {
+	ops := []string{"+", "-", "*", "/", "==", "!=", ">", "<", ">=", "<="}
+	labelSets := [][]string{{}, {"pod"}, {"machine"}, {"pod", "machine"}, {"app"}, {"app", "pod"}}
+	includes := [][]string{nil, {"pod"}, {"machine"}, {"pod", "machine"}}
+	rng := rand.New(rand.NewSource(20261012))
+	steps := []float64{100, 160, 220}
+	compared, errored, skipped := 0, 0, 0
+	byError := map[string]int{}
+	for i := 0; i < 6000; i++ {
+		leftRef, leftBody := genOperand(rng, steps)
+		rightRef, rightBody := genOperand(rng, steps)
+		m := refMatching{on: rng.Intn(2) == 0, labels: labelSets[rng.Intn(len(labelSets))]}
+		switch rng.Intn(4) {
+		case 0:
+			m.card, m.include = "group_left", includes[rng.Intn(len(includes))]
+		case 1:
+			m.card, m.include = "group_right", includes[rng.Intn(len(includes))]
+		}
+		op, returnBool := ops[rng.Intn(len(ops))], rng.Intn(2) == 0
+		if op == "+" || op == "-" || op == "*" || op == "/" {
+			returnBool = false
+		}
+		var vm *translator.VectorMatchInfo
+		if i%5 == 4 {
+			// A join with no modifier: the evaluator receives no matching info.
+			m = refMatching{}
+		} else {
+			vm = &translator.VectorMatchInfo{}
+			if m.on {
+				vm.MatchOn, vm.On = true, m.labels
+			} else {
+				vm.Ignoring = m.labels
+			}
+			if m.card != "" {
+				vm.GroupSide = m.card
+				if m.card == "group_left" {
+					vm.GroupLeft = m.include
+				} else {
+					vm.GroupRight = m.include
+				}
+			}
+		}
+		desc := fmt.Sprintf("%d: %s on=%v labels=%v card=%s include=%v bool=%v\nleft=%s\nright=%s", i, op, m.on, m.labels, m.card, m.include, returnBool, leftBody, rightBody)
+
+		want, wantErr := refMatrix(op, m, returnBool, leftRef, rightRef)
+		if wantErr == nil {
+			collided := false
+			for _, ts := range steps {
+				if samples, _ := refStep(op, m, returnBool, leftRef[ts], rightRef[ts]); refCollides(samples) {
+					collided = true
+				}
+			}
+			if collided {
+				skipped++
+				continue
+			}
+		}
+		body, err := matchBinaryMetricResults(leftBody, rightBody, op, "matrix", vm, returnBool)
+		switch {
+		case wantErr != nil:
+			errored++
+			byError[strings.SplitN(wantErr.Error(), ";", 2)[0]]++
+			if err == nil || err.Error() != wantErr.Error() {
+				t.Fatalf("error = %v, want %q\n%s", err, wantErr, desc)
+			}
+		case err != nil:
+			t.Fatalf("unexpected error %q\n%s", err, desc)
+		default:
+			compared++
+			got := decodeMatchResult(t, body)
+			if len(want) == 0 && len(got) == 0 {
+				continue
+			}
+			if !reflect.DeepEqual(got, want) {
+				t.Fatalf("got %s\nwant %v\n%s", body, want, desc)
+			}
+		}
+	}
+	t.Logf("compared %d results, %d errors %v, skipped %d collisions", compared, errored, byError, skipped)
+	if compared < 500 || errored < 500 {
+		t.Fatalf("generator no longer covers both outcomes: %d results, %d errors", compared, errored)
+	}
+}
diff --git a/internal/proxy/vector_matching.go b/internal/proxy/vector_matching.go
index c3b2ee72a..ec3e63ee5 100644
--- a/internal/proxy/vector_matching.go
+++ b/internal/proxy/vector_matching.go
@@ -325,103 +325,132 @@ func validateVectorMatchCardinality(leftBody, rightBody []byte, onLabels []strin
 }
 
 func validateVectorMatchCardinalityContext(ctx context.Context, leftBody, rightBody []byte, onLabels []string, ignoringLabels []string, allowGroupLeft, allowGroupRight bool) error {
+	_, err := firstVectorMatchFailure(ctx, leftBody, rightBody, onLabels, ignoringLabels, allowGroupLeft, allowGroupRight)
+	return err
+}
+
+// vectorMatchCounts counts, for every evaluation time, how many series of a
+// response share each match key: the non-empty on() labels, or every non-empty
+// label but the ignoring() ones.
+func vectorMatchCounts(ctx context.Context, body []byte, onLabels, ignoringLabels []string) (map[float64]map[string]int, error) {
+	var response struct {
+		Data struct {
+			Result []struct {
+				Metric map[string]string `json:"metric"`
+				Value  []interface{}     `json:"value"`
+				Values [][]interface{}   `json:"values"`
+			} `json:"result"`
+		} `json:"data"`
+	}
+	result := make(map[float64]map[string]int)
 	if err := ctx.Err(); err != nil {
-		return err
-	}
-	// Loki checks each evaluation independently. Streams which never overlap
-	// must not create a false many-to-one error across the whole range.
-	type evaluationKey struct {
-		labels string
-		time   float64
-	}
-	var countErr error
-	counts := func(body []byte) map[evaluationKey]int {
-		var response struct {
-			Data struct {
-				Result []struct {
-					Metric map[string]string `json:"metric"`
-					Value  []interface{}     `json:"value"`
-					Values [][]interface{}   `json:"values"`
-				} `json:"result"`
-			} `json:"data"`
-		}
-		result := make(map[evaluationKey]int)
-		if countErr = ctx.Err(); countErr != nil {
-			return result
-		}
-		if json.Unmarshal(body, &response) != nil {
-			return result
-		}
-		ignore := make(map[string]bool, len(ignoringLabels))
-		for _, label := range ignoringLabels {
-			ignore[label] = true
-		}
-		for _, series := range response.Data.Result {
-			if countErr = ctx.Err(); countErr != nil {
-				return result
-			}
-			labels := make(map[string]string, len(series.Metric))
-			if onLabels != nil {
-				for _, label := range onLabels {
-					if value := series.Metric[label]; value != "" {
-						labels[label] = value
-					}
+		return result, err
+	}
+	if json.Unmarshal(body, &response) != nil {
+		return result, nil
+	}
+	ignore := make(map[string]bool, len(ignoringLabels))
+	for _, label := range ignoringLabels {
+		ignore[label] = true
+	}
+	for _, series := range response.Data.Result {
+		if err := ctx.Err(); err != nil {
+			return result, err
+		}
+		labels := make(map[string]string, len(series.Metric))
+		if onLabels != nil {
+			for _, label := range onLabels {
+				if value := series.Metric[label]; value != "" {
+					labels[label] = value
 				}
-			} else {
-				for label, value := range series.Metric {
-					if !ignore[label] && value != "" {
-						labels[label] = value
-					}
+			}
+		} else {
+			for label, value := range series.Metric {
+				if !ignore[label] && value != "" {
+					labels[label] = value
 				}
 			}
-			encoded, _ := json.Marshal(labels)
-			key := string(encoded)
-			values := series.Values
-			if len(series.Value) == 2 {
-				values = append(values, series.Value)
+		}
+		encoded, _ := json.Marshal(labels)
+		key := string(encoded)
+		values := series.Values
+		if len(series.Value) == 2 {
+			values = append(values, series.Value)
+		}
+		for _, value := range values {
+			if err := ctx.Err(); err != nil {
+				return result, err
 			}
-			for _, value := range values {
-				if countErr = ctx.Err(); countErr != nil {
-					return result
-				}
-				if len(value) != 2 {
-					continue
-				}
-				if ts, ok := value[0].(float64); ok {
-					result[evaluationKey{key, ts}]++
+			if len(value) != 2 {
+				continue
+			}
+			if ts, ok := value[0].(float64); ok {
+				if result[ts] == nil {
+					result[ts] = make(map[string]int)
 				}
+				result[ts][key]++
 			}
 		}
-		return result
 	}
-	leftCounts, rightCounts := counts(leftBody), counts(rightBody)
-	if countErr != nil {
-		return countErr
+	return result, nil
+}
+
+// firstVectorMatchFailure returns the first evaluation, by time, whose join
+// Loki rejects for its cardinality, with the error it raises there. Any other
+// error (a cancelled context) is returned with no evaluation.
+func firstVectorMatchFailure(ctx context.Context, leftBody, rightBody []byte, onLabels []string, ignoringLabels []string, allowGroupLeft, allowGroupRight bool) (float64, error) {
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
+	// Loki joins one evaluation at a time and stops at the first one that
+	// fails (pkg/logql/engine.go reads the evaluator's error after every
+	// step), so the checks run per evaluation, oldest first. Streams which
+	// never overlap must not create a false many-to-one error across the
+	// whole range.
+	leftCounts, err := vectorMatchCounts(ctx, leftBody, onLabels, ignoringLabels)
+	if err != nil {
+		return 0, err
+	}
+	rightCounts, err := vectorMatchCounts(ctx, rightBody, onLabels, ignoringLabels)
+	if err != nil {
+		return 0, err
+	}
+	axis := make([]float64, 0, len(leftCounts)+len(rightCounts))
+	for ts := range leftCounts {
+		axis = append(axis, ts)
+	}
+	for ts := range rightCounts {
+		if _, both := leftCounts[ts]; !both {
+			axis = append(axis, ts)
+		}
 	}
+	sort.Float64s(axis)
 	// The "one" side must be unique even without a matching sample.
 	oneSide, side := rightCounts, "right"
 	if allowGroupRight {
 		oneSide, side = leftCounts, "left"
 	}
-	for _, count := range oneSide {
+	for _, ts := range axis {
 		if err := ctx.Err(); err != nil {
-			return err
+			return 0, err
 		}
-		if count > 1 {
-			return vectorMatchError(fmt.Sprintf("found duplicate series on the %s hand-side; many-to-many matching not allowed: matching labels must be unique on one side", side))
-		}
-	}
-	if !allowGroupLeft && !allowGroupRight {
-		for key, count := range leftCounts {
-			if err := ctx.Err(); err != nil {
-				return err
+		for _, count := range oneSide[ts] {
+			if count > 1 {
+				// Loki's text joins the two halves with ";" and no space
+				// (pkg/logql/evaluator.go:974-975).
+				return ts, vectorMatchError(fmt.Sprintf("found duplicate series on the %s hand-side;many-to-many matching not allowed: matching labels must be unique on one side", side))
 			}
-			if count > 1 && rightCounts[key] > 0 {
-				return vectorMatchError("multiple matches for labels: many-to-one matching must be explicit (group_left/group_right)")
+		}
+		if allowGroupLeft || allowGroupRight {
+			continue
+		}
+		for key, count := range leftCounts[ts] {
+			if count > 1 && rightCounts[ts][key] > 0 {
+				return ts, vectorMatchError("multiple matches for labels: many-to-one matching must be explicit (group_left/group_right)")
 			}
 		}
 	}
-	return nil
+	return 0, nil
 }
 
 // applyOnMatching joins two metric results by a specified label subset.
diff --git a/test/e2e-compat/binary_ignoring_matching_test.go b/test/e2e-compat/binary_ignoring_matching_test.go
new file mode 100644
index 000000000..f5f268fa5
--- /dev/null
+++ b/test/e2e-compat/binary_ignoring_matching_test.go
@@ -0,0 +1,226 @@
+//go:build e2e
+
+package e2e_compat
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+	"net/url"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+type ignoringFixture struct {
+	absent, present string // service_name of the streams without and with a machine label
+	evaluation      time.Time
+}
+
+func ingestIgnoringStream(t *testing.T, ts time.Time, labels map[string]string, lines int) {
+	t.Helper()
+	fields := make([]string, 0, len(labels))
+	for name := range labels {
+		fields = append(fields, name)
+	}
+	sort.Strings(fields)
+	values := make([][]string, 0, lines)
+	for i := 0; i < lines; i++ {
+		at := ts.Add(time.Duration(i) * time.Millisecond)
+		line := fmt.Sprintf("binary-ignoring-%d", i)
+		row := map[string]string{"_time": at.UTC().Format(time.RFC3339Nano), "_msg": line}
+		for name, value := range labels {
+			row[name] = value
+		}
+		encoded, _ := json.Marshal(row)
+		status, body := hardeningRequest(t, "POST", vlURL+"/insert/jsonline?_stream_fields="+strings.Join(fields, ","), string(encoded)+"\n", map[string]string{"Content-Type": "application/stream+json"})
+		if status != 200 {
+			t.Fatalf("VL ingest %d %s", status, body)
+		}
+		values = append(values, []string{strconv.FormatInt(at.UnixNano(), 10), line})
+	}
+	payload, _ := json.Marshal(map[string]any{"streams": []any{map[string]any{"stream": labels, "values": values}}})
+	status, body := hardeningRequest(t, "POST", lokiURL+"/loki/api/v1/push", string(payload), map[string]string{"Content-Type": "application/json"})
+	if status != 204 {
+		t.Fatalf("Loki ingest %d %s", status, body)
+	}
+}
+
+// ensureIgnoringFixture ingests two data sets into both stores: one service
+// whose streams carry cluster only (the label named by ignoring() is absent
+// on both operands) and one whose streams also carry a machine label (the
+// right operand then holds several series per cluster).
+func ensureIgnoringFixture(t *testing.T) ignoringFixture {
+	t.Helper()
+	suffix := time.Now().UnixNano()
+	fx := ignoringFixture{
+		absent:     fmt.Sprintf("binary-ignoring-absent-%d", suffix),
+		present:    fmt.Sprintf("binary-ignoring-present-%d", suffix),
+		evaluation: time.Now().Add(-4 * time.Minute).Truncate(time.Minute),
+	}
+	stamp := fx.evaluation.Add(-30 * time.Second)
+	ingestIgnoringStream(t, stamp, map[string]string{"service_name": fx.absent, "cluster": "c1"}, 3)
+	ingestIgnoringStream(t, stamp, map[string]string{"service_name": fx.absent, "cluster": "c2"}, 1)
+	ingestIgnoringStream(t, stamp, map[string]string{"service_name": fx.present, "cluster": "c1", "machine": "m1"}, 1)
+	ingestIgnoringStream(t, stamp, map[string]string{"service_name": fx.present, "cluster": "c1", "machine": "m2"}, 2)
+	hardeningRequest(t, "POST", vlURL+"/internal/force_flush", "", nil)
+	waitForLokiMetricDataSelector(t, `{service_name="`+fx.absent+`"}`)
+	waitForLokiMetricDataSelector(t, `{service_name="`+fx.present+`"}`)
+	return fx
+}
+
+type ignoringAnswer struct {
+	status int
+	series map[string][]string // labels -> values
+	error  string
+}
+
+func ignoringQuery(t *testing.T, base string, fx ignoringFixture, query string, instant bool) ignoringAnswer {
+	t.Helper()
+	start := fx.evaluation.UTC().Format(time.RFC3339Nano)
+	endpoint, params := "query_range", url.Values{"query": {query}, "start": {start}, "end": {fx.evaluation.Add(time.Minute).UTC().Format(time.RFC3339Nano)}, "step": {"60"}}
+	if instant {
+		endpoint, params = "query", url.Values{"query": {query}, "time": {start}}
+	}
+	status, body := hardeningRequest(t, "GET", base+"/loki/api/v1/"+endpoint+"?"+params.Encode(), "", map[string]string{"X-Scope-OrgID": "0", "X-Loki-Response-Encoding-Flags": "categorize-labels"})
+	answer := ignoringAnswer{status: status, series: map[string][]string{}}
+	if status != 200 {
+		answer.error = strings.TrimSpace(string(body))
+		var failure struct{ Message, Error string }
+		if json.Unmarshal(body, &failure) == nil {
+			if answer.error = failure.Message; answer.error == "" {
+				answer.error = failure.Error
+			}
+		}
+		return answer
+	}
+	var response struct {
+		Data struct {
+			Result []struct {
+				Metric map[string]string
+				Value  []any
+				Values [][]any
+			}
+		}
+	}
+	if err := json.Unmarshal(body, &response); err != nil {
+		t.Fatalf("%s: %v: %s", base, err, body)
+	}
+	for _, series := range response.Data.Result {
+		points := series.Values
+		if instant {
+			points = [][]any{series.Value}
+		}
+		key, _ := json.Marshal(series.Metric)
+		for _, point := range points {
+			answer.series[string(key)] = append(answer.series[string(key)], fmt.Sprintf("%v=%v", point[0], point[1]))
+		}
+	}
+	return answer
+}
+
+type ignoringCase struct {
+	name, query string
+	wantSeries  int // series Loki must return (0: Loki must fail)
+	wantError   string
+}
+
+// TestCompat_BinaryIgnoringAbsentLabel: `+ ignoring(machine)` between two sums
+// joins on the labels the sides share, whether the ignored label is absent on
+// both operands (one series per cluster holding the sum of both sides), on one
+// side, or on the right operand only with several values (Loki fails the join
+// with its duplicate-series text).
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range, loki_api_v1_query
+func TestCompat_BinaryIgnoringAbsentLabel(t *testing.T) {
+	fx := ensureIgnoringFixture(t)
+	rate := func(service string) string { return `rate({service_name="` + service + `"}[5m])` }
+	runIgnoringCases(t, fx, []ignoringCase{
+		{"top_level_join",
+			`sum by (cluster) (` + rate(fx.absent) + `) + ignoring(machine) sum by (cluster, machine) (` + rate(fx.absent) + `)`, 2, ""},
+		{"ignored_label_on_the_left_only",
+			`sum by (cluster, machine) (` + rate(fx.absent) + `) + ignoring(machine) sum by (cluster) (` + rate(fx.absent) + `)`, 2, ""},
+		{"comparison_filter",
+			`sum by (cluster) (` + rate(fx.absent) + `) > ignoring(machine) sum by (cluster, machine) (` + rate(fx.absent) + `) / 2`, 2, ""},
+		{"comparison_bool",
+			`sum by (cluster) (` + rate(fx.absent) + `) == bool ignoring(machine) sum by (cluster, machine) (` + rate(fx.absent) + `)`, 2, ""},
+		{"group_left",
+			`sum by (cluster, machine) (` + rate(fx.present) + `) / ignoring(machine) group_left sum by (cluster) (` + rate(fx.present) + `)`, 2, ""},
+		{"right_operand_holds_several_per_cluster",
+			`sum by (cluster) (` + rate(fx.present) + `) + ignoring(machine) sum by (cluster, machine) (` + rate(fx.present) + `)`, 0,
+			"found duplicate series on the right hand-side;many-to-many matching not allowed: matching labels must be unique on one side"},
+		{"left_operand_holds_several_per_cluster",
+			`sum by (cluster, machine) (` + rate(fx.present) + `) + ignoring(machine) sum by (cluster) (` + rate(fx.present) + `)`, 0,
+			"multiple matches for labels: many-to-one matching must be explicit (group_left/group_right)"},
+	})
+}
+
+// TestCompat_BinaryIgnoringOuterSum: the registry row's shape, the join under
+// an outer `sum by (cluster)`, on streams without the ignored label. It stays
+// red until an outer aggregation over a binary expression is evaluated from both
+// operands (the proxy answers the first operand's raw count today).
+// conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range, loki_api_v1_query
+func TestCompat_BinaryIgnoringOuterSum(t *testing.T) {
+	fx := ensureIgnoringFixture(t)
+	rate := `rate({service_name="` + fx.absent + `"}[5m])`
+	runIgnoringCases(t, fx, []ignoringCase{
+		{"row_shape", `sum by (cluster) (sum by (cluster) (` + rate + `) + ignoring(machine) sum by (cluster, machine) (` + rate + `))`, 2, ""},
+		{"ungrouped_outer_sum", `sum(sum by (cluster) (` + rate + `) + ignoring(machine) sum by (cluster, machine) (` + rate + `))`, 1, ""},
+	})
+}
+
+func runIgnoringCases(t *testing.T, fx ignoringFixture, cases []ignoringCase) {
+	t.Helper()
+	for _, tc := range cases {
+		for _, instant := range []bool{false, true} {
+			name := tc.name + "/range"
+			if instant {
+				name = tc.name + "/instant"
+			}
+			t.Run(name, func(t *testing.T) {
+				var loki ignoringAnswer
+				for deadline := time.Now().Add(90 * time.Second); ; {
+					// Loki answers a fresh stack with empty 200s for a while.
+					loki = ignoringQuery(t, lokiURL, fx, tc.query, instant)
+					if len(loki.series) == tc.wantSeries || tc.wantSeries == 0 && loki.status != 200 || time.Now().After(deadline) {
+						break
+					}
+					time.Sleep(500 * time.Millisecond)
+				}
+				if tc.wantSeries > 0 && len(loki.series) != tc.wantSeries {
+					t.Fatalf("Loki sanity: %d series, want %d: %+v", len(loki.series), tc.wantSeries, loki)
+				}
+				if tc.wantError != "" && (loki.status == 200 || !strings.Contains(loki.error, tc.wantError)) {
+					t.Fatalf("Loki sanity: status %d error %q, want %q", loki.status, loki.error, tc.wantError)
+				}
+				proxy := ignoringQuery(t, proxyURL, fx, tc.query, instant)
+				if proxy.status != loki.status || proxy.error != loki.error {
+					t.Fatalf("status/error: proxy %d %q, Loki %d %q", proxy.status, proxy.error, loki.status, loki.error)
+				}
+				if len(proxy.series) != len(loki.series) {
+					t.Fatalf("series: proxy %v, Loki %v", proxy.series, loki.series)
+				}
+				for key, want := range loki.series {
+					got := proxy.series[key]
+					if len(got) != len(want) {
+						t.Fatalf("series %s: proxy %v, Loki %v", key, got, want)
+					}
+					for i := range want {
+						if !ignoringPointsEqual(got[i], want[i]) {
+							t.Fatalf("series %s point %d: proxy %s, Loki %s", key, i, got[i], want[i])
+						}
+					}
+				}
+			})
+		}
+	}
+}
+
+func ignoringPointsEqual(a, b string) bool {
+	at, av, _ := strings.Cut(a, "=")
+	bt, bv, _ := strings.Cut(b, "=")
+	x, errA := strconv.ParseFloat(av, 64)
+	y, errB := strconv.ParseFloat(bv, 64)
+	return at == bt && errA == nil && errB == nil && math.Abs(x-y) <= 1e-9*math.Max(1, math.Abs(y))
+}

From 72af131b3cc4ff9ab4111189f89b3aa6f660ff84 Mon Sep 17 00:00:00 2001
From: szibis <szibis@users.noreply.github.com>
Date: Mon, 12 Oct 2026 13:13:38 +0200
Subject: [PATCH 2/4] chore(conformance): regenerate the parity reports for
 binary-ignoring-matching

---
 conformance/registry/generated/wiring.json | 29 +++++++++++++++++++---
 conformance/reports/parity-gaps.md         |  2 +-
 2 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/conformance/registry/generated/wiring.json b/conformance/registry/generated/wiring.json
index 84533f221..d89a0eaf5 100644
--- a/conformance/registry/generated/wiring.json
+++ b/conformance/registry/generated/wiring.json
@@ -181,11 +181,13 @@
       "test/e2e-compat/metadata_scan_admission_test.go"
     ],
     "loki_api_v1_query": [
+      "internal/proxy/binary_ignoring_matching_test.go",
       "internal/proxy/parenthesized_range_test.go",
       "internal/proxy/series_limit_routes_test.go",
       "internal/proxy/tenant_query_limits_test.go",
       "internal/proxy/tumbling_axis_parity_test.go",
       "internal/proxy/unwrap_conversion_error_test.go",
+      "test/e2e-compat/binary_ignoring_matching_test.go",
       "test/e2e-compat/extracted_suffix_collision_test.go",
       "test/e2e-compat/parenthesized_range_or_filters_test.go",
       "test/e2e-compat/parser_capture_identity_test.go",
@@ -196,6 +198,8 @@
     ],
     "loki_api_v1_query_range": [
       "internal/proxy/backend_heavy_test.go",
+      "internal/proxy/binary_ignoring_matching_test.go",
+      "internal/proxy/binary_matching_differential_test.go",
       "internal/proxy/drilldown_breakdown_exact_test.go",
       "internal/proxy/drilldown_breakdown_matrix_test.go",
       "internal/proxy/extracted_suffix_collision_test.go",
@@ -210,6 +214,7 @@
       "internal/proxy/unwrap_conversion_error_test.go",
       "internal/translator/extracted_suffix_test.go",
       "internal/translator/regexp_escaping_test.go",
+      "test/e2e-compat/binary_ignoring_matching_test.go",
       "test/e2e-compat/drilldown_field_breakdown_test.go",
       "test/e2e-compat/extracted_suffix_collision_test.go",
       "test/e2e-compat/heavy_query_protection_test.go",
@@ -361,6 +366,11 @@
     "semantics/backend-timeout-argument": [
       "internal/proxy/backend_heavy_test.go"
     ],
+    "semantics/binary-ignoring-matching": [
+      "internal/proxy/binary_ignoring_matching_test.go",
+      "internal/proxy/binary_matching_differential_test.go",
+      "test/e2e-compat/binary_ignoring_matching_test.go"
+    ],
     "semantics/client-cancellation-propagates": [
       "internal/proxy/backend_heavy_test.go"
     ],
@@ -592,6 +602,15 @@
       "operator-configurable-limits",
       "semantics/backend-aborted-response"
     ],
+    "internal/proxy/binary_ignoring_matching_test.go": [
+      "loki_api_v1_query",
+      "loki_api_v1_query_range",
+      "semantics/binary-ignoring-matching"
+    ],
+    "internal/proxy/binary_matching_differential_test.go": [
+      "loki_api_v1_query_range",
+      "semantics/binary-ignoring-matching"
+    ],
     "internal/proxy/compat_cache_freshness_test.go": [
       "semantics/metadata-answers-include-last-24h-like-loki"
     ],
@@ -895,6 +914,11 @@
     "test/e2e-compat/bad_request_status_parity_test.go": [
       "loki-compatible-profile"
     ],
+    "test/e2e-compat/binary_ignoring_matching_test.go": [
+      "loki_api_v1_query",
+      "loki_api_v1_query_range",
+      "semantics/binary-ignoring-matching"
+    ],
     "test/e2e-compat/compat_option_matrix_test.go": [
       "loki-compatible-profile",
       "profiles/detected-fields-dotted-json-keys",
@@ -1064,7 +1088,7 @@
       "profiles/structured-metadata-keys-per-profile"
     ]
   },
-  "tests_with_markers": 96,
+  "tests_with_markers": 99,
   "unknown_ids": {},
   "unwired_items": [
     "api_prom_label",
@@ -1131,7 +1155,6 @@
     "quality/labels-backfill-into-cached-bucket",
     "quality/stats-volume-freshness",
     "semantics/absent-over-time-semantics",
-    "semantics/binary-ignoring-matching",
     "semantics/count-values-unsupported",
     "semantics/detected-fields-extracted-suffix",
     "semantics/drop-keep-plain-name-after-parser",
@@ -1172,5 +1195,5 @@
     "versions/capabilities-floor-vs-matrix",
     "versions/rejected-stats-query-wording-vl-1.47"
   ],
-  "wired_items": 106
+  "wired_items": 107
 }
diff --git a/conformance/reports/parity-gaps.md b/conformance/reports/parity-gaps.md
index 47d3231db..0048086c0 100644
--- a/conformance/reports/parity-gaps.md
+++ b/conformance/reports/parity-gaps.md
@@ -30,7 +30,7 @@ Last run: 2026-10-05T20:16:15Z on `fcaccfcc`, window [1791220800, 1791224400].
 | 9 | `profiles/json-extraction-list-exposes-only-listed` — Extraction lists (`\| json msg="msg"`, `\| logfmt op`) expose level from the line (open) | open | explore-visible | 18 | 1 | S | internal/proxy stage-field exposure (extraction lists) | pkg/logql/log/parser.go:671-700 JSONExpressionParser.Process (listed ids only) | TestCompat_JSONExtractionListExposesOnlyListed (test/e2e-compat, group semantics) |
 | 10 | `semantics/first-last-over-time-translation` — first_over_time and last_over_time with offset or a line filter fail in VictoriaLogs (open) | open | explore-visible | 16 | 10 | S | internal/translator first/last_over_time emission (row_min/row_max with offset and filters) | pkg/logql/first_last_over_time.go | TestCompat_FirstLastOverTimeOffsetFilter (test/e2e-compat, group semantics) |
 | 11 | `semantics/tail-history-start-limit` — Native tail ignores start and limit: Grafana Live shows no history where Loki shows its last 100 lines (open) | open | explore-visible | 15 | 1 | M | internal/proxy/tail.go openNativeTailStream (history replay before the live stream) | pkg/loghttp/tail.go:70-94 ParseTailQuery (limit, start); querier tailer sends history first | TestCompat_TailHistoryStartLimit (test/e2e-compat, group tail-multitenancy) + @explore-tail Playwright |
-| 12 | `semantics/binary-ignoring-matching` — `+ ignoring(label)` between two sums joins the wrong series (open) | open | explore-visible | 10 | 2 | S | internal/proxy/vector_matching.go (ignoring with a label absent on both sides) | pkg/logql/evaluator.go binary vector matching (ignoring) | TestCompat_BinaryIgnoringAbsentLabel (test/e2e-compat, group semantics) |
+| 12 | `semantics/binary-ignoring-matching` — `+ ignoring(label)` between two sums joins the wrong series (open) | open | explore-visible | 10 | 2 | S | translator and handleQueryRange/handleQuery: an outer aggregation over a binary expression (the binary evaluator itself matches Loki) | pkg/logql/evaluator.go:791-1128 newBinOpStepEvaluator, matchingSignature, vectorBinop, resultMetric; pkg/logql/engine.go step loop | TestCompat_BinaryIgnoringOuterSum (the row's shape, test/e2e-compat, group semantics) and TestCompat_BinaryIgnoringAbsentLabel (the join ... |
 | 13 | `semantics/parse-error-labels-on-log-lines` — A line a parser rejects carries __error__ and __error_details__ in Loki's log answer (open) | open | explore-visible | 10 | 2 | M | internal/proxy log-row assembly (stage-field exposure) + translator parse-error predicate | pkg/logql/log/parser.go:97 addErrLabel(errJSON, err) -> :777-787 SetErr/SetErrorDetails; pkg/logql/log/error.go:5 errJSON | TestCompat_ParseErrorLabelsOnLogLines (test/e2e-compat, group semantics) |
 | 14 | `semantics/absent-over-time-semantics` — absent_over_time answers when data exists and stays empty when it is absent, after a parser (open) | open | explore-visible | 8 | 6 | S | internal/translator absent_over_time (parser pipelines) | pkg/logql/evaluator.go:647-664, :1320 absentLabels | TestCompat_AbsentOverTimeWithParser (test/e2e-compat, group semantics) |
 | 15 | `semantics/stddev-stdvar-grouped-values` — stddev and stdvar by a label compute different values (open) | open | explore-visible | 7 | 5 | S | internal/translator stddev/stdvar outer aggregation (by-group fold) | pkg/logql/evaluator.go VectorAggEvaluator stddev/stdvar | TestCompat_StddevStdvarBy (test/e2e-compat, group semantics) |

From 3c9ac5c4baa6d50f0fac34b122ce82f17311e776 Mon Sep 17 00:00:00 2001
From: szibis <szibis@users.noreply.github.com>
Date: Mon, 12 Oct 2026 13:13:44 +0200
Subject: [PATCH 3/4] docs(changelog): name #742

---
 CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ee0dfba39..58877c4bf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -28,7 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   vectorBinop, matchingSignature, resultMetric and MergeBinOp, an edge table for
   `ignoring` over a label absent on both sides, on one side and on the right
   only, and handler tests on range and instant queries; the live comparison with
-  Loki is `TestCompat_BinaryIgnoringAbsentLabel` (#TBD). Not changed: the query
+  Loki is `TestCompat_BinaryIgnoringAbsentLabel` (#742). Not changed: the query
   from the conformance row `semantics/binary-ignoring-matching`, `sum by
   (cluster) (sum by (cluster) (rate(..)) + ignoring(machine) sum by (cluster,
   machine) (rate(..)))`, still answers the first operand's raw count (about 150

From 6c2ff79774d408cbdcf279d925ccbfc8ec1342b6 Mon Sep 17 00:00:00 2001
From: szibis <szibis@users.noreply.github.com>
Date: Mon, 12 Oct 2026 14:33:54 +0200
Subject: [PATCH 4/4] test: skip the outer-sum ignoring e2e until the
 aggregation over a binary lands

---
 test/e2e-compat/binary_ignoring_matching_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/test/e2e-compat/binary_ignoring_matching_test.go b/test/e2e-compat/binary_ignoring_matching_test.go
index f5f268fa5..2a771ba0c 100644
--- a/test/e2e-compat/binary_ignoring_matching_test.go
+++ b/test/e2e-compat/binary_ignoring_matching_test.go
@@ -162,6 +162,7 @@ func TestCompat_BinaryIgnoringAbsentLabel(t *testing.T) {
 // operands (the proxy answers the first operand's raw count today).
 // conformance: semantics/binary-ignoring-matching, loki_api_v1_query_range, loki_api_v1_query
 func TestCompat_BinaryIgnoringOuterSum(t *testing.T) {
+	t.Skip("needs the aggregation over a binary expression to evaluate both operands (semantics/aggregation-over-binary-dropped, #741); unskip when it lands")
 	fx := ensureIgnoringFixture(t)
 	rate := `rate({service_name="` + fx.absent + `"}[5m])`
 	runIgnoringCases(t, fx, []ignoringCase{