test(query-engine): cover temporal anchors and topk ordering

promql-compliance/HOW_TO.md

@@ -154,6 +154,33 @@ queries:
 If tolerance is omitted, values are compared exactly. A tolerance should be
 small and justified; a broad tolerance can hide a correctness bug.
 
+### Require PromQL instant-vector ordering
+
+Vector and matrix comparisons are order-insensitive by default. Opt into the
+PromQL ordering rule for an instant `topk` or `bottomk` response with:
+
+```yaml
+comparison:
+  instant_vector_order:
+    direction: descending
+```
+
+Use `ascending` for `bottomk`. Grouped TopK/BottomK keeps ordering inside a
+bucket but does not impose an order on buckets, so declare the grouping used
+by the query:
+
+```yaml
+comparison:
+  instant_vector_order:
+    direction: descending
+    grouping:
+      mode: by
+      labels: [job]
+```
+
+`mode: without` is also supported. This policy applies only to `/query`
+responses; `/query_range` and range-at-t parity remain order-insensitive.
+
 ## Run with DEBUG logging
 
 The default Compose stack uses `INFO`. Temporarily change the query engine

promql-compliance/datasets/topk-ordering.yaml

@@ -0,0 +1,20 @@
+name: topk-ordering
+series:
+  - metric: ordered_data
+    labels: {job: backend, instance: a}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]
+  - metric: ordered_data
+    labels: {job: backend, instance: b}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]
+  - metric: ordered_data
+    labels: {job: backend, instance: c}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]
+  - metric: ordered_data
+    labels: {job: frontend, instance: a}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]
+  - metric: ordered_data
+    labels: {job: frontend, instance: b}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]
+  - metric: ordered_data
+    labels: {job: frontend, instance: c}
+    samples: [{offset_seconds: 0, value: 1}, {offset_seconds: 300, value: 1}, {offset_seconds: 600, value: 1}]

promql-compliance/runner/Makefile

@@ -12,6 +12,7 @@ CI_CASES := \
 	single-rate-off-grid-rate:../datasets/single-rate.yaml:../suites/off-grid-rate.yaml \
 	aggregations-native-dag:../datasets/aggregations.yaml:../suites/native-dag-aggregations.yaml \
 	aggregations:../datasets/aggregations.yaml:../suites/aggregations.yaml \
+	topk-ordering:../datasets/topk-ordering.yaml:../suites/topk-ordering.yaml \
 	quantiles:../datasets/quantiles.yaml:../suites/quantiles.yaml
 
 # Tracks planner coverage (#738); most queries are expected to fail.

promql-compliance/runner/checked_in_fixture_validation_test.go

@@ -10,6 +10,7 @@ import (
 func TestCheckedInFixturesAndSuites(t *testing.T) {
 	pairs := []struct{ dataset, suite string }{
 		{"../datasets/aggregations.yaml", "../suites/aggregations.yaml"},
+		{"../datasets/topk-ordering.yaml", "../suites/topk-ordering.yaml"},
 		{"../datasets/quantiles.yaml", "../suites/quantiles.yaml"},
 		{"../datasets/olly-bench.yaml", "../suites/olly-bench.yaml"},
 	}

promql-compliance/runner/compare.go

@@ -24,7 +24,7 @@ type QueryAPI interface {
 type QueryReport struct {
 	Name            string              `json:"name"`
 	Expr            string              `json:"expr"`
-	Tolerance       ComparisonPolicy    `json:"tolerance"`
+	Comparison      ComparisonPolicy    `json:"comparison"`
 	Range           *ComparisonOutcome  `json:"range,omitempty"`
 	Instant         []InstantComparison `json:"instant,omitempty"`
 	ReferenceParity []ParityComparison  `json:"referenceParity,omitempty"`
@@ -55,12 +55,12 @@ type ParityComparison struct {
 // It compares the two targets and also checks range-at-t against
 // instant-at-t within each target.
 func CompareQuery(ctx context.Context, reference, test QueryAPI, query QueryCase, base time.Time, defaults ComparisonPolicy) (QueryReport, error) {
-	effective := query.EffectiveTolerance(defaults)
+	effective := query.EffectiveComparison(defaults)
 	report := QueryReport{
-		Name:      query.Name,
-		Expr:      query.Expr,
-		Tolerance: effective,
-		Passed:    true,
+		Name:       query.Name,
+		Expr:       query.Expr,
+		Comparison: effective,
+		Passed:     true,
 	}
 
 	var referenceRange, testRange model.Value
@@ -81,7 +81,7 @@ func CompareQuery(ctx context.Context, reference, test QueryAPI, query QueryCase
 	for index, instantTime := range instantTimes {
 		referenceInstant, _, referenceErr := reference.Query(ctx, query.Expr, instantTime)
 		testInstant, _, testErr := test.Query(ctx, query.Expr, instantTime)
-		outcome := responseComparison(referenceInstant, testInstant, referenceErr, testErr, effective)
+		outcome := instantResponseComparison(referenceInstant, testInstant, referenceErr, testErr, effective)
 		report.Instant = append(report.Instant, InstantComparison{
 			OffsetSeconds: query.InstantOffsetsSeconds[index],
 			Time:          instantTime,
@@ -125,6 +125,18 @@ func (q QueryCase) RangeAt(base time.Time) (clientv1.Range, error) {
 }
 
 func responseComparison(reference, test model.Value, referenceErr, testErr error, tolerance ComparisonPolicy) ComparisonOutcome {
+	return responseComparisonWith(reference, test, referenceErr, testErr, func(reference, test model.Value) string {
+		return compareValues(reference, test, tolerance)
+	})
+}
+
+func instantResponseComparison(reference, test model.Value, referenceErr, testErr error, policy ComparisonPolicy) ComparisonOutcome {
+	return responseComparisonWith(reference, test, referenceErr, testErr, func(reference, test model.Value) string {
+		return compareInstantValues(reference, test, policy)
+	})
+}
+
+func responseComparisonWith(reference, test model.Value, referenceErr, testErr error, compare func(model.Value, model.Value) string) ComparisonOutcome {
 	outcome := ComparisonOutcome{}
 	if referenceErr != nil {
 		outcome.ReferenceError = referenceErr.Error()
@@ -136,7 +148,7 @@ func responseComparison(reference, test model.Value, referenceErr, testErr error
 		outcome.Passed = false
 		return outcome
 	}
-	outcome.Diff = compareValues(reference, test, tolerance)
+	outcome.Diff = compare(reference, test)
 	outcome.Passed = outcome.Diff == ""
 	return outcome
 }
@@ -166,6 +178,122 @@ func compareValues(reference, test model.Value, tolerance ComparisonPolicy) stri
 	return compareNormalized(referenceNormalized, testNormalized, tolerance)
 }
 
+func compareInstantValues(reference, test model.Value, policy ComparisonPolicy) string {
+	if policy.InstantVectorOrder == nil {
+		return compareValues(reference, test, policy)
+	}
+	referenceVector, referenceIsVector := reference.(model.Vector)
+	testVector, testIsVector := test.(model.Vector)
+	if !referenceIsVector || !testIsVector {
+		return fmt.Sprintf(
+			"instant vector order policy requires vector results, got reference %T and test %T",
+			reference,
+			test,
+		)
+	}
+
+	referenceGroups, err := orderedVectorGroups(referenceVector, policy.InstantVectorOrder.Grouping)
+	if err != nil {
+		return orderedVectorDiff("reference instant vector grouping is invalid: "+err.Error(), referenceVector, testVector)
+	}
+	testGroups, err := orderedVectorGroups(testVector, policy.InstantVectorOrder.Grouping)
+	if err != nil {
+		return orderedVectorDiff("test instant vector grouping is invalid: "+err.Error(), referenceVector, testVector)
+	}
+	if len(referenceGroups) != len(testGroups) {
+		return orderedVectorDiff(fmt.Sprintf("instant vector ordered group count differs: reference %d, test %d", len(referenceGroups), len(testGroups)), referenceVector, testVector)
+	}
+	for key, referenceGroup := range referenceGroups {
+		testGroup, found := testGroups[key]
+		if !found {
+			return orderedVectorDiff(fmt.Sprintf("instant vector ordered group %q is absent from test result", key), referenceVector, testVector)
+		}
+		if err := validateOrderedValues(testGroup, policy.InstantVectorOrder.Direction); err != nil {
+			return orderedVectorDiff(fmt.Sprintf("test instant vector ordered group %q: %v", key, err), referenceVector, testVector)
+		}
+		if len(referenceGroup) != len(testGroup) {
+			return orderedVectorDiff(fmt.Sprintf("instant vector ordered group %q sample count differs: reference %d, test %d", key, len(referenceGroup), len(testGroup)), referenceVector, testVector)
+		}
+		if diff := compareSampleMembership(referenceGroup, testGroup, policy.ValueTolerance); diff != "" {
+			return orderedVectorDiff(fmt.Sprintf("instant vector ordered group %q %s", key, diff), referenceVector, testVector)
+		}
+	}
+	return ""
+}
+
+func compareSampleMembership(reference, test []normalizedSample, tolerance *Tolerance) string {
+	referenceSamples := append([]normalizedSample(nil), reference...)
+	testSamples := append([]normalizedSample(nil), test...)
+	sortNormalizedSamples(referenceSamples)
+	sortNormalizedSamples(testSamples)
+	return compareNormalized(
+		normalizedValue{Type: "vector", Samples: referenceSamples},
+		normalizedValue{Type: "vector", Samples: testSamples},
+		ComparisonPolicy{ValueTolerance: tolerance},
+	)
+}
+
+func orderedVectorDiff(reason string, reference, test model.Vector) string {
+	return fmt.Sprintf("%s\nreference: %v\ntest: %v", reason, reference, test)
+}
+
+func orderedVectorGroups(vector model.Vector, grouping *OrderGrouping) (map[string][]normalizedSample, error) {
+	groups := make(map[string][]normalizedSample)
+	var currentKey string
+	hasCurrentGroup := false
+	for _, sample := range vector {
+		key := orderGroupKey(sample.Metric, grouping)
+		if !hasCurrentGroup || key != currentKey {
+			if _, seen := groups[key]; seen {
+				return nil, fmt.Errorf("group %q is not contiguous", key)
+			}
+			groups[key] = []normalizedSample{}
+			currentKey = key
+			hasCurrentGroup = true
+		}
+		groups[key] = append(groups[key], normalizedSample{Metric: metricString(sample.Metric), Timestamp: int64(sample.Timestamp), Value: float64(sample.Value)})
+	}
+	return groups, nil
+}
+
+func validateOrderedValues(samples []normalizedSample, direction string) error {
+	for index := 1; index < len(samples); index++ {
+		previous, current := samples[index-1], samples[index]
+		if direction == instantOrderDescending && previous.Value < current.Value {
+			return fmt.Errorf("values are not descending at sample %d", index)
+		}
+		if direction == instantOrderAscending && previous.Value > current.Value {
+			return fmt.Errorf("values are not ascending at sample %d", index)
+		}
+	}
+	return nil
+}
+
+func orderGroupKey(metric model.Metric, grouping *OrderGrouping) string {
+	if grouping == nil {
+		return ""
+	}
+	labels := make(model.Metric)
+	if grouping.Mode == orderGroupingBy {
+		for _, label := range grouping.Labels {
+			labelName := model.LabelName(label)
+			labels[labelName] = metric[labelName]
+		}
+		return metricString(labels)
+	}
+	excluded := make(map[model.LabelName]struct{}, len(grouping.Labels))
+	excluded[model.MetricNameLabel] = struct{}{}
+	for _, label := range grouping.Labels {
+		excluded[model.LabelName(label)] = struct{}{}
+	}
+	for label, value := range metric {
+		if _, skip := excluded[label]; !skip {
+			labels[label] = value
+		}
+	}
+	return metricString(labels)
+}
+
 type normalizedValue struct {
 	Type    string             `json:"type"`
 	Samples []normalizedSample `json:"samples,omitempty"`