MQE: info() derive info series @ reference only when the first argument's vector paths are uniform

CHANGELOG.md

@@ -32,7 +32,7 @@
 * [BUGFIX] Compactor, Store-gateway: Fix the store-gateway always logging `num_series=0` in its `loaded new block` message. #16276
 * [BUGFIX] Ingest storage: Account for protobuf framing when splitting Remote Write 1.0 requests so generated Kafka record data stays within `-ingest-storage.kafka.producer-max-record-size-bytes` when individual series and metadata entries fit. #16160
 * [BUGFIX] Memcached: Don't close connections to caches on well-formed server errors. #16303
-* [BUGFIX] MQE: Propagate an `@` modifier or offset from the `info` function's first argument to its info series matchers, matching Prometheus. #16220
+* [BUGFIX] MQE: Propagate an `@` modifier or offset from the `info` function's first argument to its info series matchers, matching Prometheus. #16220 #16497
 * [BUGFIX] MQE: Fix an issue where series were joined in binary operations using the wrong labels when `group_left()`/`group_right()` were used in combination with `ignoring()`. This bug manifested as valid queries returning an error `grouping labels must ensure unique matches`. #16387
 * [BUGFIX] MQE: Fix queries and rules containing a subquery whose range is shorter than its step (e.g. `foo[10m:3d]`) failing with `last bucket must not be before first bucket`. #16442
 * [BUGFIX] MQE: Fix `avg_over_time()` over native histograms losing precision when experimental range vector splitting is enabled. The Kahan compensation of each split range was discarded instead of being carried over to the other split ranges. #16472

pkg/streamingpromql/planning/core/function_call.go

@@ -77,13 +77,12 @@ func MaterializeFunctionCall(ctx context.Context, f *FunctionCall, materializer
 	}
 
 	if f.Function == functions.FUNCTION_INFO && len(f.Args) == 2 {
-		// Propagate the @/offset modifiers of the first selector in the first argument to the
-		// data label selector, mirroring Prometheus's infoSelectHints, so that info series are
-		// selected at the same (shifted) time as the samples they enrich. This is derived from
-		// the plan here rather than in the operator factory because operators do not support
-		// generic child traversal, so nested selectors would not be reachable there.
+		// Pin the data label selector to the first argument's @/offset (mirroring Prometheus's
+		// infoSelectHints), but only when the vector paths share one reference; otherwise leave the
+		// per-step default. Done here, not in the operator factory, where nested selectors aren't
+		// reachable.
 		if dataLabelSelector, ok := children[1].(*functions.DataLabelSelector); ok {
-			if ts, offset, found := infoSelectTimestampAndOffset(f.Args[0]); found {
+			if ts, offset, uniform := infoSelectTimestampAndOffset(f.Args[0]); uniform {
 				dataLabelSelector.Selector.Timestamp = TimestampFromTime(ts)
 				dataLabelSelector.Selector.Offset = offset.Milliseconds()
 			}
@@ -133,10 +132,13 @@ func (f *FunctionCall) QueriedTimeRange(queryTimeRange types.QueryTimeRange, loo
 }
 
 // infoSeriesQueriedTimeRange returns the time range over which the info function selects its info
-// series, using the @ timestamp and offset derived from the first argument. When the first argument
-// has no such reference, this matches the data label selector's own evaluation-time range.
+// series. Like MaterializeFunctionCall, it pins to the first argument's @/offset only when the
+// vector paths share one reference, so the advertised range matches the operator's fetch.
 func (f *FunctionCall) infoSeriesQueriedTimeRange(queryTimeRange types.QueryTimeRange, lookbackDelta time.Duration) planning.QueriedTimeRange {
-	ts, offset, _ := infoSelectTimestampAndOffset(f.Args[0])
+	ts, offset, uniform := infoSelectTimestampAndOffset(f.Args[0])
+	if !uniform {
+		ts, offset = nil, 0
+	}
 	minT, maxT := selectors.ComputeQueriedTimeRange(queryTimeRange, TimestampFromTime(ts), 0, offset.Milliseconds(), lookbackDelta, false, false)
 	return planning.NewQueriedTimeRange(timestamp.Time(minT), timestamp.Time(maxT))
 }

pkg/streamingpromql/planning/core/info.go

@@ -22,37 +22,92 @@ type DataLabelSelector struct {
 	*DataLabelSelectorDetails
 }
 
-// infoSelectTimestampAndOffset returns the @ timestamp and offset modifiers the info function's
-// data label selector should use, given the plan subtree of the info function's first argument.
-//
-// It mirrors Prometheus's infoSelectHints: the reference time is derived from the first vector or
-// matrix selector found in a pre-order traversal of node, which matches the first VectorSelector
-// found by parser.Inspect over the equivalent expression AST. (In the AST the @/offset modifiers of
-// a range vector selector live on its inner VectorSelector, but in the plan they live on the
-// MatrixSelector node itself, so both node types are considered here.) Enclosing subqueries shift
-// the selector's reference time: their offsets add up until an @ timestamp anchors it, after which
-// modifiers further out are irrelevant.
-func infoSelectTimestampAndOffset(node planning.Node) (ts *time.Time, offset time.Duration, found bool) {
+// infoSelectTimestampAndOffset returns the @ timestamp and offset the info data label selector
+// should use, taken from the first vector or matrix selector in a pre-order traversal of the info
+// function's first argument (mirroring Prometheus's infoSelectHints; enclosing subqueries compose
+// their offsets until an @ timestamp anchors the reference). uniform is false when the vector paths
+// don't all share one reference (mixed references, selector-free vectors, or only scalar selectors),
+// in which case the caller must not pin the lookup so enrichment stays step-dependent.
+func infoSelectTimestampAndOffset(node planning.Node) (ts *time.Time, offset time.Duration, uniform bool) {
+	w := infoReferenceWalk{uniform: true}
+	w.inspect(node, nil)
+	if !w.found {
+		return nil, 0, false
+	}
+	return w.first.timestamp, w.first.offset, w.uniform && !w.referenceFree
+}
+
+type infoSeriesReference struct {
+	timestamp *time.Time
+	offset    time.Duration
+}
+
+func (r infoSeriesReference) equal(other infoSeriesReference) bool {
+	if r.timestamp == nil || other.timestamp == nil {
+		return r.timestamp == nil && other.timestamp == nil && r.offset == other.offset
+	}
+	return r.timestamp.Add(-r.offset).Equal(other.timestamp.Add(-other.offset))
+}
+
+type infoReferenceWalk struct {
+	first         infoSeriesReference
+	found         bool
+	referenceFree bool
+	uniform       bool
+}
+
+// inspect records the effective reference of each selector in a vector-producing subtree and
+// returns whether it contains any. enclosingSubqueries are those it's nested within, outermost first.
+func (w *infoReferenceWalk) inspect(node planning.Node, enclosingSubqueries []*Subquery) bool {
+	if valueType, err := node.ResultType(); err != nil || (valueType != parser.ValueTypeVector && valueType != parser.ValueTypeMatrix) {
+		// Not a vector-producing data source for enrichment.
+		return false
+	}
+
 	switch n := node.(type) {
 	case *VectorSelector:
-		return n.Timestamp, n.Offset, true
+		w.addSelector(n.Timestamp, n.Offset, enclosingSubqueries)
+		return true
 	case *MatrixSelector:
-		return n.Timestamp, n.Offset, true
+		w.addSelector(n.Timestamp, n.Offset, enclosingSubqueries)
+		return true
+	case *DataLabelSelector:
+		// A nested info's data label selector is selector syntax, not an evaluated vector.
+		return false
 	}
 
+	if sq, ok := node.(*Subquery); ok {
+		enclosingSubqueries = append(enclosingSubqueries, sq)
+	}
+
+	hasSelector := false
 	for child := range planning.ChildrenIter(node) {
-		if ts, offset, found := infoSelectTimestampAndOffset(child); found {
-			// As the traversal unwinds back through an enclosing subquery, compose its modifiers:
-			// add its offset and take its @ timestamp, unless the reference time is already anchored.
-			if sq, ok := node.(*Subquery); ok && ts == nil {
-				offset += sq.Offset
-				ts = sq.Timestamp
-			}
-			return ts, offset, found
+		if w.inspect(child, enclosingSubqueries) {
+			hasSelector = true
 		}
 	}
+	if !hasSelector {
+		// A selector-free vector follows evaluation time; it can't be pinned by another path.
+		w.referenceFree = true
+	}
+	return hasSelector
+}
+
+// addSelector records a selector's reference, composing its enclosing subqueries' offsets and
+// innermost @ timestamp.
+func (w *infoReferenceWalk) addSelector(ts *time.Time, offset time.Duration, enclosingSubqueries []*Subquery) {
+	ref := infoSeriesReference{timestamp: ts, offset: offset}
+	for i := len(enclosingSubqueries) - 1; ref.timestamp == nil && i >= 0; i-- {
+		ref.offset += enclosingSubqueries[i].Offset
+		ref.timestamp = enclosingSubqueries[i].Timestamp
+	}
 
-	return nil, 0, false
+	if !w.found {
+		w.first = ref
+		w.found = true
+	} else if !w.first.equal(ref) {
+		w.uniform = false
+	}
 }
 
 func (t *DataLabelSelector) Details() proto.Message {

pkg/streamingpromql/planning_test.go

@@ -2263,40 +2263,48 @@ func (t *versioningTestNode) MinimumRequiredPlanVersion(types.QueryTimeRange) (p
 // info series' lookback window must be contributed by the info data label selector. Previously it
 // was not, so the overall queried time range - which drives remote-execution data fetching - could
 // miss the info series and silently drop enrichment.
-func TestInfoQueriedTimeRangeCoversPinnedTime(t *testing.T) {
-	const (
-		pinnedMs int64 = 120_000 // metric @ 120
-		rangeMs  int64 = 60_000  // metric[1m]
-	)
+func TestInfoQueriedTimeRange(t *testing.T) {
 	lookbackDelta := 5 * time.Minute
+	// Evaluate well after any pinned time so a range pinned to a selector doesn't reach it.
+	evalTime := timestamp.Time(0).Add(40 * time.Minute)
+
+	// The +1ms on MinT excludes the sample exactly one lookback delta before (see ComputeQueriedTimeRange).
+	testCases := map[string]struct {
+		expr         string
+		expectedMinT time.Time
+		expectedMaxT time.Time
+	}{
+		// A uniform reference pins the lookup to the shared time; the range selector alone queries
+		// no lookback, so the info series' lookback window must come from the data label selector.
+		"uniform reference pins to the shared time": {
+			expr:         "info(last_over_time(metric[1m] @ 120))",
+			expectedMinT: timestamp.Time(120_000 - lookbackDelta.Milliseconds() + 1),
+			expectedMaxT: timestamp.Time(120_000),
+		},
+		// Selectors pinned at different times aren't uniform: info series are matched per step, so
+		// the range must reach the evaluation time rather than be capped at the pinned times.
+		"non-uniform references use evaluation time": {
+			expr:         "info(metric @ 120 or other_metric @ 180)",
+			expectedMinT: timestamp.Time(120_000 - lookbackDelta.Milliseconds() + 1),
+			expectedMaxT: evalTime,
+		},
+	}
 
 	opts := NewTestEngineOpts()
 	planner, err := NewQueryPlanner(opts, NewMaximumSupportedVersionQueryPlanVersionProvider())
 	require.NoError(t, err)
-
-	// Evaluate well after the pinned time so the (buggy) evaluation-time info window does not
-	// happen to overlap the pinned window.
-	evalTime := timestamp.Time(0).Add(40 * time.Minute)
 	queryTimeRange := types.NewInstantQueryTimeRange(evalTime)
 
-	plan, err := planner.NewQueryPlan(context.Background(), "info(last_over_time(metric[1m] @ 120))", queryTimeRange, lookbackDelta, false, NoopPlanningObserver{})
-	require.NoError(t, err)
-
-	queried, err := plan.Root.QueriedTimeRange(queryTimeRange, lookbackDelta)
-	require.NoError(t, err)
-
-	// The info series must be fetched across their lookback window as of the pinned time. The +1ms
-	// on the lower bound excludes the sample exactly one lookback delta before (see
-	// selectors.ComputeQueriedTimeRange), matching how a pinned instant selector is queried.
-	expectedMinT := timestamp.Time(pinnedMs - lookbackDelta.Milliseconds() + 1)
-	expectedMaxT := timestamp.Time(pinnedMs)
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			plan, err := planner.NewQueryPlan(context.Background(), tc.expr, queryTimeRange, lookbackDelta, false, NoopPlanningObserver{})
+			require.NoError(t, err)
 
-	require.Equal(t, expectedMinT, queried.MinT, "queried range must reach back to cover the info series lookback window at the pinned time")
-	require.Equal(t, expectedMaxT, queried.MaxT)
+			queried, err := plan.Root.QueriedTimeRange(queryTimeRange, lookbackDelta)
+			require.NoError(t, err)
 
-	// Sanity check that the range selector alone would not have covered this: its queried start is
-	// only pinnedTime-range (no lookback), so without the fix the info series before that point are
-	// missed.
-	rangeSelectorMinT := timestamp.Time(pinnedMs - rangeMs + 1)
-	require.True(t, expectedMinT.Before(rangeSelectorMinT), "test setup: info lookback window must extend before the range selector's queried start")
+			require.Equal(t, tc.expectedMinT, queried.MinT)
+			require.Equal(t, tc.expectedMaxT, queried.MaxT)
+		})
+	}
 }

pkg/streamingpromql/testdata/upstream/info.test

@@ -294,24 +294,21 @@ eval range from 8m to 9m step 1m info(last_over_time(metric[5m:1m] offset 6m))
 # Modifiers in scalar-only subexpressions do not shift enrichment for the vector operand.
 # Reordering scalar-vector multiplication therefore produces the same current info labels.
 # Multiplication drops the metric name, which info() must not restore.
-# Unsupported by streaming engine.
-# eval instant at 8m info(scalar(metric @ 120) * metric)
-#     {instance="a", job="1", version="new"} 16
+eval instant at 8m info(scalar(metric @ 120) * metric)
+    {instance="a", job="1", version="new"} 16
 
 eval instant at 8m info(metric * scalar(metric @ 120))
     {instance="a", job="1", version="new"} 16
 
-# Unsupported by streaming engine.
-# eval instant at 8m info(scalar(metric offset 6m) * metric)
-#     {instance="a", job="1", version="new"} 16
+eval instant at 8m info(scalar(metric offset 6m) * metric)
+    {instance="a", job="1", version="new"} 16
 
 # Different reference times on vector-producing operands are ambiguous: there is no single
 # shared reference, so info series are selected and matched at each evaluation step instead of
 # being pinned to the first selector's reference.
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(metric @ 120 or other_metric)
-#     metric{instance="a", job="1", version="new"} 2 2
-#     other_metric{instance="b", job="1", version="new"} 8 8
+eval range from 8m to 9m step 1m info(metric @ 120 or other_metric)
+    metric{instance="a", job="1", version="new"} 2 2
+    other_metric{instance="b", job="1", version="new"} 8 8
 
 # Operand order within the first argument must not affect which info series are selected.
 eval range from 8m to 9m step 1m info(other_metric or metric @ 120)
@@ -320,22 +317,19 @@ eval range from 8m to 9m step 1m info(other_metric or metric @ 120)
 
 # When all vector inputs are fixed at different times, info() still uses evaluation-time
 # matching. It must remain step-dependent so shared timestamps don't depend on the range start.
-# Unsupported by streaming engine.
-# eval range from 4m to 9m step 1m info(metric @ 120 or other_metric @ 480)
-#     metric{instance="a", job="1", version="new"} 2 2 2 2 2 2
-#     other_metric{instance="b", job="1", version="new"} 8 8 8 8 8 8
+eval range from 4m to 9m step 1m info(metric @ 120 or other_metric @ 480)
+    metric{instance="a", job="1", version="new"} 2 2 2 2 2 2
+    other_metric{instance="b", job="1", version="new"} 8 8 8 8 8 8
 
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480)
-#     metric{instance="a", job="1", version="new"} 2 2
-#     other_metric{instance="b", job="1", version="new"} 8 8
+eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480)
+    metric{instance="a", job="1", version="new"} 2 2
+    other_metric{instance="b", job="1", version="new"} 8 8
 
 # The optional data label selector is syntax rather than an evaluated vector. Its modifiers
 # must not make preprocessing wrap it and change the concrete type expected by info().
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480, {version=~".*"} @ 120)
-#     metric{instance="a", job="1", version="new"} 2 2
-#     other_metric{instance="b", job="1", version="new"} 8 8
+eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480, {version=~".*"} @ 120)
+    metric{instance="a", job="1", version="new"} 2 2
+    other_metric{instance="b", job="1", version="new"} 8 8
 
 # A selector-free vector has no pinned reference for info series. Enrichment must therefore
 # follow each evaluation step instead of being fixed to the range start.
@@ -349,15 +343,13 @@ eval range from 4m to 8m step 1m info(label_replace(label_replace(vector(1), "in
 # A fixed selector in another vector-producing branch does not provide a shared reference for
 # selector-free vectors. It must not pin their info series selection or enrichment to its
 # historical timestamp, regardless of operand order.
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(label_replace(label_replace(vector(1), "instance", "a", "__name__", ".*"), "job", "1", "__name__", ".*") or other_metric @ 120, {version=~".*"})
-#     {instance="a", job="1", version="new"} 1 1
-#     other_metric{instance="b", job="1", version="new"} 2 2
+eval range from 8m to 9m step 1m info(label_replace(label_replace(vector(1), "instance", "a", "__name__", ".*"), "job", "1", "__name__", ".*") or other_metric @ 120, {version=~".*"})
+    {instance="a", job="1", version="new"} 1 1
+    other_metric{instance="b", job="1", version="new"} 2 2
 
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(other_metric @ 120 or label_replace(label_replace(vector(1), "instance", "a", "__name__", ".*"), "job", "1", "__name__", ".*"), {version=~".*"})
-#     {instance="a", job="1", version="new"} 1 1
-#     other_metric{instance="b", job="1", version="new"} 2 2
+eval range from 8m to 9m step 1m info(other_metric @ 120 or label_replace(label_replace(vector(1), "instance", "a", "__name__", ".*"), "job", "1", "__name__", ".*"), {version=~".*"})
+    {instance="a", job="1", version="new"} 1 1
+    other_metric{instance="b", job="1", version="new"} 2 2
 
 # Syntactically different fixed modifiers that resolve to the same lookup time stay uniform.
 eval range from 8m to 9m step 1m info(metric @ 180 offset 1m or other_metric @ 120)
@@ -366,10 +358,9 @@ eval range from 8m to 9m step 1m info(metric @ 180 offset 1m or other_metric @ 1
 
 # A nested info data-label selector is selector syntax, not a vector-producing data source. The
 # nested selector-free input also does not share the sibling selector's historical reference.
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(info(vector(1), {foo=~".*"}) or metric @ 120)
-#     {} 1 1
-#     metric{instance="a", job="1", version="new"} 2 2
+eval range from 8m to 9m step 1m info(info(vector(1), {foo=~".*"}) or metric @ 120)
+    {} 1 1
+    metric{instance="a", job="1", version="new"} 2 2
 
 clear
 
@@ -383,19 +374,17 @@ load 1m
     target_info{instance="b", job="1", version="old"} _ _ _ _ _ _ _ _ 1 stale
     target_info{instance="b", job="1", version="new"} _ _ _ _ _ _ _ _ _ 1
 
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480)
-#     metric{instance="a", job="1", version="old"} 2 _
-#     metric{instance="a", job="1", version="new"} _ 2
-#     other_metric{instance="b", job="1", version="old"} 8 _
-#     other_metric{instance="b", job="1", version="new"} _ 8
-
-# Unsupported by streaming engine.
-# eval range from 8m to 9m step 1m info(other_metric @ 480 or metric @ 120)
-#     metric{instance="a", job="1", version="old"} 2 _
-#     metric{instance="a", job="1", version="new"} _ 2
-#     other_metric{instance="b", job="1", version="old"} 8 _
-#     other_metric{instance="b", job="1", version="new"} _ 8
+eval range from 8m to 9m step 1m info(metric @ 120 or other_metric @ 480)
+    metric{instance="a", job="1", version="old"} 2 _
+    metric{instance="a", job="1", version="new"} _ 2
+    other_metric{instance="b", job="1", version="old"} 8 _
+    other_metric{instance="b", job="1", version="new"} _ 8
+
+eval range from 8m to 9m step 1m info(other_metric @ 480 or metric @ 120)
+    metric{instance="a", job="1", version="old"} 2 _
+    metric{instance="a", job="1", version="new"} _ 2
+    other_metric{instance="b", job="1", version="old"} 8 _
+    other_metric{instance="b", job="1", version="new"} _ 8
 
 clear