fix(promql): keep value-field grouping labels out of matching filter propagation (#9242)

* fix(promql): keep value-field grouping labels out of matching filter propagation

#9202 propagates matching-label matchers to the scanned selector using the
planned operands' tag columns. For an aggregated operand those are its
grouping labels, and agg_modifier_to_col resolves by(...) names against the
input schema only, so a value field named in by(...) is reported as a tag.
A value field varies between the samples of one series, so lowering a
matcher on it below sample selection (PromInstantManipulate) can drop the
newest sample, promote a stale one from the lookback window, and fabricate
a match the un-rewritten query does not produce.

Track the by(...) labels that name value fields of the aggregated operand's
input in PromPlannerContext::aggregation_field_labels, and refuse to
propagate matchers on them while still requiring a matcher to name a
grouping label to cross an aggregation.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test: sync matching_filter result fixture comment with the PR number

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/query/src/promql/planner.rs

@@ -174,6 +174,15 @@ struct PromPlannerContext {
     time_index_column: Option<String>,
     field_columns: Vec<String>,
     tag_columns: Vec<String>,
+    /// `by(...)` labels of the aggregation that produced this operand that are not series tags of
+    /// its input, i.e. value fields (or a label an inner aggregation already reported as one).
+    ///
+    /// An aggregation reports every `by(...)` label it finds in its input schema among its tag
+    /// columns, and a value field named there is a group key of the aggregate rather than a
+    /// property of a series: its value varies between the samples of one series. Lowering a
+    /// matcher on it into the scan would change which samples are selected (#9242), so
+    /// [`matching_filters`] refuses to propagate such a matcher.
+    aggregation_field_labels: Vec<String>,
     /// Use metric engine internal series identifier column (`__tsid`) as series key.
     ///
     /// This is enabled only when the underlying scan can provide `__tsid` (`UInt64`). The planner
@@ -1359,6 +1368,8 @@ impl PromPlanner {
             // lhs is a literal, rhs is a column
             (Some(mut expr), None) => {
                 let input = self.prom_expr_to_plan(rhs, query_engine_state).await?;
+                // Arithmetic against a literal preserves the series labels, so
+                // `aggregation_field_labels` passes through with `tag_columns` unchanged.
                 // check if the literal is a special time expr
                 if let Some(time_expr) = self.try_build_special_time_expr_with_context(lhs) {
                     expr = time_expr
@@ -1515,6 +1526,8 @@ impl PromPlanner {
                     binary_expr,
                     &left_context.tag_columns,
                     &right_context.tag_columns,
+                    &left_context.aggregation_field_labels,
+                    &right_context.aggregation_field_labels,
                 ) {
                     // A copied matcher belongs to the scan, not to the operand's identity:
                     // `absent()` turns `selector_matcher` into the labels it reports, so the
@@ -2207,6 +2220,7 @@ impl PromPlanner {
             self.ctx.time_index_column = Some(SPECIAL_TIME_FUNCTION.to_string());
             self.ctx.reset_table_name_and_schema();
             self.ctx.tag_columns = vec![];
+            self.ctx.aggregation_field_labels.clear();
             self.ctx.field_columns = vec![DEFAULT_FIELD_COLUMN.to_string()];
             LogicalPlan::Extension(Extension {
                 node: Arc::new(
@@ -2599,11 +2613,27 @@ impl PromPlanner {
             None => {
                 if update_ctx {
                     self.ctx.tag_columns.clear();
+                    self.ctx.aggregation_field_labels.clear();
                 }
                 Ok(vec![self.create_time_index_column_expr()?])
             }
             Some(LabelModifier::Include(labels)) => {
                 if update_ctx {
+                    // A `by(...)` label can name a value field of the input instead of a tag. The
+                    // aggregate still reports it among its tag columns below, but unlike a tag it
+                    // is a group key rather than a property of a series: its value varies between
+                    // the samples of one series, so a matcher on it must stay above sample
+                    // selection (#9242). Record it, before the tag columns are overwritten.
+                    self.ctx.aggregation_field_labels = labels
+                        .labels
+                        .iter()
+                        .filter(|label| {
+                            self.ctx.field_columns.contains(label)
+                                || !self.ctx.tag_columns.contains(label)
+                                || self.ctx.aggregation_field_labels.contains(label)
+                        })
+                        .cloned()
+                        .collect();
                     self.ctx.tag_columns.clear();
                 }
                 let mut exprs = Vec::with_capacity(labels.labels.len());
@@ -2655,6 +2685,12 @@ impl PromPlanner {
                 if update_ctx {
                     // change the tag columns in context
                     self.ctx.tag_columns = all_fields.iter().map(|col| (*col).clone()).collect();
+                    // `without(...)` drops the value fields of the input from its grouping labels,
+                    // so a label stays a grouping label in name only if it survives in the input
+                    // schema (e.g. an inner aggregation that grouped by it).
+                    self.ctx
+                        .aggregation_field_labels
+                        .retain(|label| all_fields.iter().any(|col| *col == label));
                 }
 
                 // collect remaining fields and convert to col expr
@@ -3348,6 +3384,8 @@ impl PromPlanner {
             .cloned()
             .collect();
         self.ctx.tag_columns = tags;
+        // The operand is a plain selector: its tag columns are the table's.
+        self.ctx.aggregation_field_labels.clear();
 
         self.ctx.use_tsid = false;
 
@@ -3360,6 +3398,7 @@ impl PromPlanner {
         self.ctx.time_index_column = Some(SPECIAL_TIME_FUNCTION.to_string());
         self.ctx.reset_table_name_and_schema();
         self.ctx.tag_columns = vec![];
+        self.ctx.aggregation_field_labels.clear();
         self.ctx.field_columns = vec![DEFAULT_FIELD_COLUMN.to_string()];
         self.ctx.use_tsid = false;
 
@@ -5204,6 +5243,7 @@ impl PromPlanner {
         self.ctx.time_index_column = Some(SPECIAL_TIME_FUNCTION.to_string());
         self.ctx.reset_table_name_and_schema();
         self.ctx.tag_columns = vec![];
+        self.ctx.aggregation_field_labels.clear();
         self.ctx.field_columns = vec![greptime_value().to_string()];
         Ok(LogicalPlan::Extension(Extension {
             node: Arc::new(
@@ -5295,6 +5335,7 @@ impl PromPlanner {
         });
         // scalar plan have no tag columns
         self.ctx.tag_columns.clear();
+        self.ctx.aggregation_field_labels.clear();
         self.ctx.field_columns.clear();
         self.ctx
             .field_columns
@@ -5364,6 +5405,9 @@ impl PromPlanner {
             ),
         });
 
+        // The absent series carries the equality matchers as labels, not the input's
+        // tags or value fields, so the input's field grouping labels no longer apply.
+        self.ctx.aggregation_field_labels.clear();
         Ok(absent_plan)
     }
 
@@ -13366,6 +13410,92 @@ Projection: count(prometheus_tsdb_head_series.greptime_value) AS my_series, prom
         plan.display_indent().to_string()
     }
 
+    /// [`build_test_table_provider_with_distinct_tags`] plus a `status` string column: a value
+    /// field that is neither a primary key nor a value column of the metric, so
+    /// `count by(status) (...)` still reports it among the aggregation's tag columns.
+    async fn build_test_table_provider_with_string_field(
+        table_tags: &[(&str, &[&str])],
+    ) -> DfTableSourceProvider {
+        let catalog_list = MemoryCatalogManager::with_default_setup();
+        for (table_name, tags) in table_tags {
+            let mut columns = tags
+                .iter()
+                .map(|tag| {
+                    ColumnSchema::new(
+                        (*tag).to_string(),
+                        ConcreteDataType::string_datatype(),
+                        false,
+                    )
+                })
+                .collect::<Vec<_>>();
+            columns.push(
+                ColumnSchema::new(
+                    greptime_timestamp().to_string(),
+                    ConcreteDataType::timestamp_millisecond_datatype(),
+                    false,
+                )
+                .with_time_index(true),
+            );
+            columns.push(ColumnSchema::new(
+                greptime_value().to_string(),
+                ConcreteDataType::float64_datatype(),
+                true,
+            ));
+            columns.push(ColumnSchema::new(
+                "status".to_string(),
+                ConcreteDataType::string_datatype(),
+                true,
+            ));
+            let table_meta = TableMetaBuilder::empty()
+                .schema(Arc::new(Schema::new(columns)))
+                .primary_key_indices((0..tags.len()).collect())
+                .next_column_id(1024)
+                .build()
+                .unwrap();
+            let table_info = TableInfoBuilder::default()
+                .name((*table_name).to_string())
+                .meta(table_meta)
+                .build()
+                .unwrap();
+
+            assert!(
+                catalog_list
+                    .register_table_sync(RegisterTableRequest {
+                        catalog: DEFAULT_CATALOG_NAME.to_string(),
+                        schema: DEFAULT_SCHEMA_NAME.to_string(),
+                        table_name: (*table_name).to_string(),
+                        table_id: 1024,
+                        table: EmptyTable::from_table_info(&table_info),
+                    })
+                    .is_ok()
+            );
+        }
+
+        DfTableSourceProvider::new(
+            catalog_list,
+            false,
+            QueryContext::arc(),
+            DummyDecoder::arc(),
+            false,
+        )
+    }
+
+    async fn build_matching_filter_plan_with_string_field(query: &str) -> String {
+        let table_provider = build_test_table_provider_with_string_field(&[
+            ("metric_a", &["host", "device"]),
+            ("metric_b", &["host", "device"]),
+        ])
+        .await;
+        let plan = PromPlanner::stmt_to_plan(
+            table_provider,
+            &build_eval_stmt(query),
+            &build_query_engine_state(),
+        )
+        .await
+        .unwrap();
+        plan.display_indent().to_string()
+    }
+
     #[tokio::test]
     async fn binary_matching_label_filter_reaches_both_operands() {
         for query in [
@@ -13405,6 +13535,27 @@ Projection: count(prometheus_tsdb_head_series.greptime_value) AS my_series, prom
         }
     }
 
+    #[tokio::test]
+    async fn binary_value_field_matcher_is_not_copied_across_aggregations() {
+        // `status` varies between the samples of one series, so filtering the other operand by it
+        // would drop the newest sample before sample selection (#9242).
+        let query = r#"count by(status) (metric_a) / on(status) count by(status) (metric_b{status="ready"})"#;
+        let plan = build_matching_filter_plan_with_string_field(query).await;
+        assert_eq!(
+            plan.matches(r#"Utf8("ready")"#).count(),
+            1,
+            "{query}\n{plan}"
+        );
+    }
+
+    #[tokio::test]
+    async fn binary_matching_label_filter_reaches_aggregations_grouping_by_tags() {
+        // `count`, not `sum`: the string value field is not summable.
+        let query = r#"count by(host) (metric_a) / on(host) count by(host) (metric_b{host="foo"})"#;
+        let plan = build_matching_filter_plan_with_string_field(query).await;
+        assert_eq!(plan.matches(r#"Utf8("foo")"#).count(), 2, "{query}\n{plan}");
+    }
+
     #[tokio::test]
     async fn binary_matching_label_filter_skips_unproven_expressions() {
         for query in [

src/query/src/promql/planner/matching_filters.rs

@@ -62,12 +62,30 @@ const LABEL_PRESERVING_RANGE_FUNCTIONS: [&str; 20] = [
 /// as well constrain a value field, which the parsed expression does not distinguish from a
 /// label. For an aggregated operand they are its grouping labels, which is what makes filtering
 /// its input equivalent to filtering its output.
+///
+/// A value field is not a series identity though: its value may vary between the samples of one
+/// series, so a matcher on it must not be lowered below sample selection (`PromInstantManipulate`,
+/// `PromSeriesDivide`), where discarding a sample promotes an older one out of the lookback window
+/// and fabricates a match. An aggregated operand's tag columns are its grouping labels rather than
+/// the tags of its input, so those labels may name value fields; a matcher therefore only crosses
+/// an aggregation when it names one of the grouping labels that aggregation partitions by, and
+/// `left_field_labels`/`right_field_labels` rule out the grouping labels the planner found to be
+/// value fields of the aggregated operand's input
+/// (`PromPlannerContext::aggregation_field_labels`).
 pub(super) fn propagate(
     binary: &BinaryExpr,
     left_tags: &[String],
     right_tags: &[String],
+    left_field_labels: &[String],
+    right_field_labels: &[String],
 ) -> Option<BinaryExpr> {
-    match try_propagate(binary, left_tags, right_tags) {
+    match try_propagate(
+        binary,
+        left_tags,
+        right_tags,
+        left_field_labels,
+        right_field_labels,
+    ) {
         Ok(rewritten) => Some(rewritten),
         Err(reason) => {
             common_telemetry::debug!("Matching filter not propagated ({reason}): {binary}");
@@ -81,6 +99,8 @@ fn try_propagate(
     binary: &BinaryExpr,
     left_tags: &[String],
     right_tags: &[String],
+    left_field_labels: &[String],
+    right_field_labels: &[String],
 ) -> Result<BinaryExpr, &'static str> {
     if !matches!(
         binary.op.id(),
@@ -101,9 +121,18 @@ fn try_propagate(
         .as_ref()
         .and_then(|modifier| modifier.matching.as_ref());
 
+    // The grouping labels of every aggregation the operands cross, collected before `rewritten`
+    // takes the mutable borrows that rule out borrowing `binary` again.
+    let left_grouped = modifier_label_names(&binary.lhs);
+    let left_ignored = modifier_excluded_label_names(&binary.lhs);
+    let right_grouped = modifier_label_names(&binary.rhs);
+    let right_ignored = modifier_excluded_label_names(&binary.rhs);
+
     let mut rewritten = binary.clone();
-    let left = selector_matchers(&mut rewritten.lhs).ok_or("left operand is not a selector")?;
-    let right = selector_matchers(&mut rewritten.rhs).ok_or("right operand is not a selector")?;
+    let (left, left_restricted) =
+        targeted_selector(&mut rewritten.lhs).ok_or("left operand is not a selector")?;
+    let (right, right_restricted) =
+        targeted_selector(&mut rewritten.rhs).ok_or("right operand is not a selector")?;
     if !left.or_matchers.is_empty() || !right.or_matchers.is_empty() {
         return Err("selector has an or matcher group");
     }
@@ -113,11 +142,16 @@ fn try_propagate(
         !name.starts_with("__")
             && left_tags.contains(name)
             && right_tags.contains(name)
+            // A grouping label that is a value field of the operand's input is not a series tag.
+            && !left_field_labels.contains(name)
+            && !right_field_labels.contains(name)
             && match matching {
                 None => true,
                 Some(LabelModifier::Include(on)) => on.labels.contains(name),
                 Some(LabelModifier::Exclude(ignoring)) => !ignoring.labels.contains(name),
             }
+            && (!left_restricted || crosses_aggregation(name, &left_grouped, &left_ignored))
+            && (!right_restricted || crosses_aggregation(name, &right_grouped, &right_ignored))
     };
     let constraints = left
         .matchers
@@ -150,9 +184,9 @@ fn matches_every_value(matcher: &Matcher) -> bool {
 /// by a grouping label drops exactly the matching output series and leaves the remaining
 /// aggregated values untouched.
 ///
-/// `topk`, `bottomk` and `limitk` rank across a group and carry the input labels through, so
-/// filtering before them changes the candidate set. `count_values` adds an output label that
-/// does not exist in its input.
+/// `topk` and `bottomk` rank across a group and carry the input labels through, so filtering
+/// before them changes the candidate set. `count_values` adds an output label that does not exist
+/// in its input.
 fn partitions_by_grouping_labels(aggregate: &AggregateExpr) -> bool {
     matches!(
         aggregate.op.id(),
@@ -168,20 +202,22 @@ fn partitions_by_grouping_labels(aggregate: &AggregateExpr) -> bool {
     )
 }
 
-/// Returns the matchers of the vector selector an operand scans, or `None` when the operand's
-/// output labels are not proven to be that selector's labels.
-fn selector_matchers(expr: &mut Expr) -> Option<&mut Matchers> {
+/// Returns the matchers of the vector selector an operand scans, together with whether the
+/// operand is restricted to the grouping labels an aggregation on its path partitions by, or
+/// `None` when the operand's output labels are not proven to be that selector's labels.
+fn targeted_selector(expr: &mut Expr) -> Option<(&mut Matchers, bool)> {
     match expr {
-        Expr::VectorSelector(selector) => Some(&mut selector.matchers),
-        Expr::Paren(paren) => selector_matchers(&mut paren.expr),
+        Expr::VectorSelector(selector) => Some((&mut selector.matchers, false)),
+        Expr::Paren(paren) => targeted_selector(&mut paren.expr),
         Expr::Aggregate(aggregate) if partitions_by_grouping_labels(aggregate) => {
-            selector_matchers(&mut aggregate.expr)
+            let (matchers, _) = targeted_selector(&mut aggregate.expr)?;
+            Some((matchers, aggregate.modifier.is_some()))
         }
         Expr::Call(call) if LABEL_PRESERVING_RANGE_FUNCTIONS.contains(&call.func.name) => {
             // Every other argument is a scalar, so position does not matter.
             let matrix = single_matrix_argument(&call.args.args)?;
             match call.args.args[matrix].as_mut() {
-                Expr::MatrixSelector(selector) => Some(&mut selector.vs.matchers),
+                Expr::MatrixSelector(selector) => Some((&mut selector.vs.matchers, false)),
                 _ => None,
             }
         }
@@ -190,6 +226,49 @@ fn selector_matchers(expr: &mut Expr) -> Option<&mut Matchers> {
     }
 }
 
+/// Collects the labels every partitioning aggregation on [`targeted_selector`]' path to the
+/// scanned selector groups by (`by(...)`, the `LabelModifier::Include` of an aggregation).
+fn modifier_label_names(expr: &Expr) -> Vec<&String> {
+    let mut grouped = Vec::new();
+    collect_modifier_label_names(expr, true, &mut grouped);
+    grouped
+}
+
+/// Collects the labels every partitioning aggregation on [`targeted_selector`]' path to the
+/// scanned selector excludes from its grouping labels (`without(...)`, the
+/// `LabelModifier::Exclude` of an aggregation).
+fn modifier_excluded_label_names(expr: &Expr) -> Vec<&String> {
+    let mut ignored = Vec::new();
+    collect_modifier_label_names(expr, false, &mut ignored);
+    ignored
+}
+
+fn collect_modifier_label_names<'a>(expr: &'a Expr, included: bool, out: &mut Vec<&'a String>) {
+    match expr {
+        Expr::Paren(paren) => collect_modifier_label_names(&paren.expr, included, out),
+        Expr::Aggregate(aggregate) if partitions_by_grouping_labels(aggregate) => {
+            match aggregate.modifier.as_ref() {
+                Some(LabelModifier::Include(labels)) if included => out.extend(&labels.labels),
+                Some(LabelModifier::Exclude(labels)) if !included => out.extend(&labels.labels),
+                _ => (),
+            }
+            collect_modifier_label_names(&aggregate.expr, included, out);
+        }
+        _ => (),
+    }
+}
+
+/// Whether an operand an aggregation partitions by `grouped` labels (with `without(...)`, by
+/// every label but `ignored`) still carries `name` as an output label, so that a matcher on
+/// `name` reaches it on both sides of that aggregation.
+fn crosses_aggregation(name: &String, grouped: &[&String], ignored: &[&String]) -> bool {
+    if grouped.is_empty() {
+        !ignored.contains(&name)
+    } else {
+        grouped.contains(&name)
+    }
+}
+
 fn single_matrix_argument(args: &[Box<Expr>]) -> Option<usize> {
     let mut found = None;
     for (index, arg) in args.iter().enumerate() {
@@ -210,11 +289,30 @@ mod tests {
     use super::*;
 
     fn rewrite_with(query: &str, left_tags: &[&str], right_tags: &[&str]) -> Option<Expr> {
+        rewrite_labels_with(query, left_tags, &[], right_tags, &[])
+    }
+
+    /// [`rewrite_with`] with the grouping labels of each operand that the planner found to be
+    /// value fields of its input rather than tags (the metric's tags are `["host", "zone"]`).
+    fn rewrite_labels_with(
+        query: &str,
+        left_tags: &[&str],
+        left_field_labels: &[&str],
+        right_tags: &[&str],
+        right_field_labels: &[&str],
+    ) -> Option<Expr> {
         let Expr::Binary(binary) = parse(query).unwrap() else {
             panic!("expected binary")
         };
         let owned = |tags: &[&str]| tags.iter().map(|tag| tag.to_string()).collect::<Vec<_>>();
-        propagate(&binary, &owned(left_tags), &owned(right_tags)).map(Expr::Binary)
+        propagate(
+            &binary,
+            &owned(left_tags),
+            &owned(right_tags),
+            &owned(left_field_labels),
+            &owned(right_field_labels),
+        )
+        .map(Expr::Binary)
     }
 
     fn rewrite(query: &str) -> Option<Expr> {
@@ -298,6 +396,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn does_not_propagate_grouping_labels_that_are_value_fields() {
+        // `status` is a value field of both metrics, though `count by(status)` reports it as a
+        // grouping label of the aggregate; its value varies between the samples of one series.
+        assert!(
+            rewrite_labels_with(
+                r#"count by(status) (a) / on(status) count by(status) (b{status="ready"})"#,
+                &["status"],
+                &["status"],
+                &["status"],
+                &["status"],
+            )
+            .is_none()
+        );
+        // `host` is a tag, `status` is a value field: only the tag may propagate.
+        assert!(
+            rewrite_labels_with(
+                r#"sum by(host, status) (a) / on(status) sum by(host, status) (b{status="ready"})"#,
+                &["host", "status"],
+                &["status"],
+                &["host", "status"],
+                &["status"],
+            )
+            .is_none()
+        );
+    }
+
+    #[test]
+    fn propagates_grouping_labels_of_aggregated_operands() {
+        assert_rewrite(
+            r#"count by(host) (a) / on(host) count by(host) (b{host="x"})"#,
+            r#"count by(host) (a{host="x"}) / on(host) count by(host) (b{host="x"})"#,
+        );
+        assert_rewrite(
+            r#"sum by(host) (rate(a[5m])) / on(host) sum by(host) (b{host="x"})"#,
+            r#"sum by(host) (rate(a{host="x"}[5m])) / on(host) sum by(host) (b{host="x"})"#,
+        );
+    }
+
+    #[test]
+    fn keeps_matchers_outside_the_grouping_labels_on_their_own_operand() {
+        assert_rewrite(
+            r#"sum by(host) (a) / on(host) sum by(host) (b{host="x",status="ready"})"#,
+            r#"sum by(host) (a{host="x"}) / on(host) sum by(host) (b{host="x",status="ready"})"#,
+        );
+        // `without(host)` leaves `host` out of the grouping labels, so nothing may cross it.
+        assert!(rewrite(r#"avg without(host) (a) / on(host) b{host="x"}"#).is_none());
+        assert_rewrite(
+            r#"avg without(zone) (a) / on(host) b{host="x"}"#,
+            r#"avg without(zone) (a{host="x"}) / on(host) b{host="x"}"#,
+        );
+    }
+
     #[test]
     fn leaves_selecting_aggregations_alone() {
         for query in [

tests/cases/standalone/common/tql/matching_filter.result

@@ -332,6 +332,59 @@ tql eval(0, 5, '5s') counter_metric / on(host) sum by(host)(gauge_metric{device=
 | host2 | 1970-01-01T00:00:05 | 20.0                                                    |
 +-------+---------------------+---------------------------------------------------------+
 
+-- A grouping label that is a value field, not a tag: its value varies between samples of one
+-- series, so a matcher on it must not filter the scan before sample selection (#9242).
+create table pr9202_a (
+    ts timestamp time index,
+    host string,
+    val double,
+    `status` string,
+    primary key (host)
+);
+
+Affected Rows: 0
+
+create table pr9202_b (
+    ts timestamp time index,
+    host string,
+    val double,
+    `status` string,
+    primary key (host)
+);
+
+Affected Rows: 0
+
+insert into pr9202_a values
+    (0, 'h1', 10, 'ready'),
+    (5000, 'h1', 20, 'busy');
+
+Affected Rows: 2
+
+insert into pr9202_b values
+    (5000, 'h1', 5, 'ready');
+
+Affected Rows: 1
+
+-- The 5s sample of pr9202_a has status="busy" and no partner; filtering it before sample
+-- selection would fall back to the stale "ready" sample and fabricate a result row.
+tql eval (5, 5, '1s', '1m')
+    count by(status) (pr9202_a)
+    / on(status)
+    count by(status) (pr9202_b{status="ready"});
+
++--------+----+-------------------------------------------------------------+-------------------------------------------------------------------+
+| status | ts | pr9202_a.count(pr9202_a.val) / pr9202_b.count(pr9202_b.val) | pr9202_a.count(pr9202_a.status) / pr9202_b.count(pr9202_b.status) |
++--------+----+-------------------------------------------------------------+-------------------------------------------------------------------+
++--------+----+-------------------------------------------------------------+-------------------------------------------------------------------+
+
+drop table pr9202_a;
+
+Affected Rows: 0
+
+drop table pr9202_b;
+
+Affected Rows: 0
+
 -- `topk` preserves its input labels, and filtering before `topk` changes the set of
 -- candidates it ranks, so the rewrite must not reach its input.
 -- SQLNESS SORT_RESULT 3 1

tests/cases/standalone/common/tql/matching_filter.sql

@@ -115,6 +115,42 @@ tql eval(0, 5, '5s') sum by(host)(counter_metric{device="eth0"}) / on(host) sum
 -- SQLNESS SORT_RESULT 3 1
 tql eval(0, 5, '5s') counter_metric / on(host) sum by(host)(gauge_metric{device="eth0"});
 
+-- A grouping label that is a value field, not a tag: its value varies between samples of one
+-- series, so a matcher on it must not filter the scan before sample selection (#9242).
+create table pr9202_a (
+    ts timestamp time index,
+    host string,
+    val double,
+    `status` string,
+    primary key (host)
+);
+
+create table pr9202_b (
+    ts timestamp time index,
+    host string,
+    val double,
+    `status` string,
+    primary key (host)
+);
+
+insert into pr9202_a values
+    (0, 'h1', 10, 'ready'),
+    (5000, 'h1', 20, 'busy');
+
+insert into pr9202_b values
+    (5000, 'h1', 5, 'ready');
+
+-- The 5s sample of pr9202_a has status="busy" and no partner; filtering it before sample
+-- selection would fall back to the stale "ready" sample and fabricate a result row.
+tql eval (5, 5, '1s', '1m')
+    count by(status) (pr9202_a)
+    / on(status)
+    count by(status) (pr9202_b{status="ready"});
+
+drop table pr9202_a;
+
+drop table pr9202_b;
+
 -- `topk` preserves its input labels, and filtering before `topk` changes the set of
 -- candidates it ranks, so the rewrite must not reach its input.
 -- SQLNESS SORT_RESULT 3 1