From 233e23b3524c9d268b7929ee2f730d48ba50f227 Mon Sep 17 00:00:00 2001 From: discord9 Date: Mon, 21 Sep 2026 07:09:31 +0000 Subject: [PATCH] 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 | 151 +++++++++++++++ .../src/promql/planner/matching_filters.rs | 179 ++++++++++++++++-- .../common/tql/matching_filter.result | 53 ++++++ .../standalone/common/tql/matching_filter.sql | 36 ++++ 4 files changed, 405 insertions(+), 14 deletions(-) diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index bea2cae586c..94081c2d284 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -174,6 +174,15 @@ struct PromPlannerContext { time_index_column: Option, field_columns: Vec, tag_columns: Vec, + /// `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, /// 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::>(); + 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 [ diff --git a/src/query/src/promql/planner/matching_filters.rs b/src/query/src/promql/planner/matching_filters.rs index 3ed5ee93baf..386d95e9228 100644 --- a/src/query/src/promql/planner/matching_filters.rs +++ b/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 { - 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 { 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]) -> Option { 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 { + 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 { let Expr::Binary(binary) = parse(query).unwrap() else { panic!("expected binary") }; let owned = |tags: &[&str]| tags.iter().map(|tag| tag.to_string()).collect::>(); - 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 { @@ -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 [ diff --git a/tests/cases/standalone/common/tql/matching_filter.result b/tests/cases/standalone/common/tql/matching_filter.result index 98834ad0902..8f5bebbe8c3 100644 --- a/tests/cases/standalone/common/tql/matching_filter.result +++ b/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 diff --git a/tests/cases/standalone/common/tql/matching_filter.sql b/tests/cases/standalone/common/tql/matching_filter.sql index fa2bc3371d5..72da3da70a0 100644 --- a/tests/cases/standalone/common/tql/matching_filter.sql +++ b/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