fix(logical): a quantile of merged panes keeps its samples' sign, so pane DDSketches compose through avg

crates/logical-asap-optimization/src/per_computation_candidates.rs

@@ -1674,8 +1674,9 @@ fn fold_stats(reduction: &Reduction, input: &Rc<OperatorNode>) -> Option<Propaga
 /// over a composed candidate. An exact sum accumulator, a renaming and a
 /// heap estimate whose update weights carry the
 /// [`WeightDomain::SamplesOfMetric`] requirement keep it, and so does a
-/// quantile estimate of those values: it is one of them, or between two of
-/// them, so it shares their sign. Anything else (rate, arithmetic, other
+/// quantile estimate of those values, built over the raw window or merged
+/// from its tumbling panes: it is one of them, or between two of them, so it
+/// shares their sign. Anything else (rate, arithmetic, other
 /// functions) ends it. Only the sign of the values is read from it.
 fn estimates_samples_of(node: &OperatorNode) -> Option<String> {
     match &node.operator {
@@ -1689,11 +1690,17 @@ fn estimates_samples_of(node: &OperatorNode) -> Option<String> {
             _ => None,
         },
         // Only a φ in [0, 1] picks one of the values; outside it the answer
-        // is -Inf or +Inf, whatever their sign.
+        // is -Inf or +Inf, whatever their sign. Merged tumbling panes of one
+        // definition ([`merged_chain`]) are read through their `PaneChain`:
+        // a quantile of the merged state is still one of the panes'
+        // samples, or between two of them.
         Operator::ASAP(ASAPOp::SummaryEstimate {
             summary_input,
             query: SketchStatistic::Quantile { q },
-        }) if (0.0..=1.0).contains(q) => match summary_input.asap()? {
+        }) if (0.0..=1.0).contains(q) => match merged_chain(summary_input)
+            .unwrap_or(summary_input)
+            .asap()?
+        {
             ASAPOp::SummaryAgg {
                 input:
                     SummaryUpdate {
@@ -1704,6 +1711,17 @@ fn estimates_samples_of(node: &OperatorNode) -> Option<String> {
                 filter: None,
                 child,
                 ..
+            }
+            | ASAPOp::PaneChain {
+                input:
+                    SummaryUpdate {
+                        item: None,
+                        weight: SummaryInputExpr::Column(ColumnRef::SampleValue),
+                        ..
+                    },
+                filter: None,
+                child,
+                ..
             } => estimates_samples_of(child),
             _ => None,
         },
@@ -4425,6 +4443,75 @@ mod tests {
         }
     }
 
+    /// A quantile read from merged tumbling panes of a metric's samples is
+    /// still one of the samples (or between two), so it carries their sign
+    /// as the same quantile over the raw window does; φ outside [0, 1]
+    /// still carries none.
+    #[test]
+    fn a_quantile_of_merged_panes_shares_its_samples_sign() {
+        let inventory = inventory("quantile_over_time(0.99, m[100s])");
+        let estimate = (0..inventory.targets.len())
+            .flat_map(|t| composed_with(&inventory, t))
+            .find_map(|(_, root)| {
+                OperatorNode::reachable(&root)
+                    .into_iter()
+                    .find(|n| matches!(n.asap(), Some(ASAPOp::SummaryEstimate { .. })))
+            })
+            .expect("a sketch estimate");
+        let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = estimate.asap() else {
+            unreachable!()
+        };
+        let Some(ASAPOp::SummaryAgg {
+            child: range,
+            family,
+            input,
+            reduction,
+            grouping,
+            filter,
+        }) = summary_input.asap()
+        else {
+            panic!("{summary_input:?}")
+        };
+        let chain = Rc::new(
+            OperatorNode::new(Operator::ASAP(ASAPOp::PaneChain {
+                child: Rc::clone(range.children()[0]),
+                family: family.clone(),
+                input: input.clone(),
+                reduction: reduction.clone(),
+                grouping: grouping.clone(),
+                filter: filter.clone(),
+                pane_width_ms: 10_000,
+                pane_count: 10,
+            }))
+            .unwrap(),
+        );
+        let merged = Rc::new(
+            OperatorNode::new(Operator::ASAP(ASAPOp::SummaryMerge {
+                children: vec![chain],
+            }))
+            .unwrap(),
+        );
+        let read = |q: f64| {
+            OperatorNode::with_schema(
+                Operator::ASAP(ASAPOp::SummaryEstimate {
+                    summary_input: Rc::clone(&merged),
+                    query: SketchStatistic::Quantile { q },
+                }),
+                estimate.schema.clone(),
+            )
+        };
+        assert_eq!(estimates_samples_of(&read(0.99)).as_deref(), Some("m"));
+        assert_eq!(
+            fold_stats(&Reduction::by(vec![]), &Rc::new(read(0.99)))
+                .unwrap()
+                .sign_statistic,
+            Some(samples_sign_statistic("m"))
+        );
+        for q in [-1.0, 2.0] {
+            assert_eq!(estimates_samples_of(&read(q)), None, "q = {q}");
+        }
+    }
+
     /// The metric whose raw samples, or plain sums of them, a composed node's
     /// values estimate: a whole-expression heap's estimates and an exact
     /// sum over the samples are; a rate or any other function is not. A

crates/planner/tests/runtime_install.rs

@@ -1217,3 +1217,163 @@ async fn every_topk_form_answers_whole_series() {
         assert_full_series_answers(&run, expected);
     }
 }
+
+/// The MVP benchmark's p99 query: the per-series p99 of a gauge over 100 s,
+/// averaged per (region, service, job), every second, within 0.05 at
+/// failure probability 0.3, over 1,000 `data` series sampled every 100 ms,
+/// with `data` declared non-negative or not.
+fn p99_avg_workload(values_non_negative: bool) -> PlanningWorkload {
+    serde_json::from_str(&format!(
+        r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{{
+              "query": "avg by (region, service, job) (quantile_over_time(0.99, data[100s]))",
+              "demand": {{"fixed_interval": 1000}},
+              "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3,
+                  "accuracy_scope": "whole_result"}}}},
+              "predictability": {{"predictable": {{"known_at": null}}}}}}]}},
+            "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 10000.0}}}},
+              "sources": {{"data": {{"series_count": 1000, "metric_type": "gauge",
+                  "sample_interval": 100, "values_non_negative": {values_non_negative}}}}}}}}}"#
+    ))
+    .unwrap()
+}
+
+/// [`p99_avg_workload`] planned for a deployment that keeps no raw samples.
+async fn plan_p99_avg(values_non_negative: bool) -> Result<InstallablePlan, String> {
+    let capabilities = DeploymentCapabilities {
+        raw_data_retained: false,
+        ..DeploymentCapabilities::default()
+    };
+    plan_installable(UserInput::new(
+        &p99_avg_workload(values_non_negative),
+        FrontendInput::Promql { histograms: None },
+        PlanningModels::builtin().with_deployment_capabilities(&capabilities),
+    ))
+    .await
+    .map_err(|error| error.to_string())
+}
+
+/// Without raw samples, the only valid plan of the p99 query keeps one
+/// DDSketch per series in one-second panes built at ingestion and merges
+/// the window's panes at query time (P5-m1). A quantile of merged panes is
+/// still one of the samples, so a declared non-negative gauge carries the
+/// sketch's relative bound through `avg`: P5-m1 is selected, its answers
+/// equal the query-time reference's and are within 5% of the exact
+/// averages.
+#[tokio::test]
+async fn a_non_negative_gauges_merged_ddsketch_panes_carry_the_relative_bound_through_avg() {
+    let plan = plan_p99_avg(true).await.expect("a valid plan");
+    let selection = plan.selection.as_ref().expect("Stage 3's report");
+    assert_eq!(selection.selected_candidate_id, "P5-m1");
+    assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion");
+    assert!(
+        plan.query_time_raw_windows.is_empty(),
+        "no raw samples read"
+    );
+    // The query's guarantee, with `data`'s declared sign and its 1,000
+    // series bound as Stage 3 binds them: a relative error of at most 0.05
+    // at failure probability at most 0.3 (DDSketch's bound is
+    // deterministic).
+    let root = plan.queries[0].root_node_id;
+    let guarantee = plan
+        .dag
+        .dag
+        .nodes
+        .iter()
+        .find(|node| node.id == root)
+        .and_then(|node| node.guarantee.clone())
+        .expect("the query's guarantee");
+    let sign =
+        asap_logical_asap_optimization::accuracy::composition::samples_sign_statistic("data");
+    let bound = guarantee.bind(&|statistic| match statistic {
+        "input_row_count" => Some(1_000.0),
+        _ => (statistic == sign).then_some(1.0),
+    });
+    assert_eq!(
+        bound.spec.accuracy_metric,
+        asap_types::ir::properties::AccuracyMetric::Relative
+    );
+    let error = bound.spec.error.evaluate().expect("a numeric bound");
+    assert!(error <= 0.05, "{error}");
+    let failure = bound
+        .spec
+        .failure_probability
+        .evaluate()
+        .expect("a numeric failure probability");
+    assert!(failure <= 0.3, "{failure}");
+
+    // Eight non-negative series, one sample every 100 ms over the last
+    // 110 s: two per (region, service, job) group.
+    let evaluation_ms = EVALUATION_MS;
+    let mut samples = Vec::new();
+    let mut per_group: BTreeMap<(String, String, String), Vec<Vec<f64>>> = BTreeMap::new();
+    for series in 0..8 {
+        let region = format!("r{}", series % 2);
+        let service = format!("s{}", series / 4);
+        let job = "j".to_string();
+        let mut window = Vec::new();
+        for step in 0..1_100i64 {
+            let timestamp_ms = evaluation_ms - 110_000 + 50 + step * 100;
+            let value = 10.0 * (series + 1) as f64 + ((step * 37 + series * 11) % 997) as f64;
+            samples.push(Sample {
+                labels: BTreeMap::from([
+                    ("__name__".to_string(), "data".to_string()),
+                    ("region".to_string(), region.clone()),
+                    ("service".to_string(), service.clone()),
+                    ("job".to_string(), job.clone()),
+                    ("instance".to_string(), format!("i{series}")),
+                ]),
+                timestamp_ms,
+                value,
+            });
+            if timestamp_ms > evaluation_ms - 100_000 {
+                window.push(value);
+            }
+        }
+        per_group
+            .entry((region, service, job))
+            .or_default()
+            .push(window);
+    }
+    let run = run_data_plane(
+        &serde_json::to_string(&plan).unwrap(),
+        &samples,
+        evaluation_ms,
+        &mut SummaryStore::new(),
+    );
+    assert_eq!(run.answers_per_query, run.reference_answers_per_query);
+    let answers = &run.answers_per_query[0];
+    assert_eq!(answers.len(), per_group.len());
+    for (labels, value) in answers {
+        let key = (
+            labels["region"].clone(),
+            labels["service"].clone(),
+            labels["job"].clone(),
+        );
+        // The exact answer: the mean of the group's per-series p99s, each
+        // PromQL's interpolated quantile of the series' window.
+        let exact = per_group[&key]
+            .iter()
+            .map(|window| {
+                let mut sorted = window.clone();
+                sorted.sort_by(f64::total_cmp);
+                let rank = 0.99 * (sorted.len() - 1) as f64;
+                let (low, high) = (rank.floor() as usize, rank.ceil() as usize);
+                sorted[low] + (sorted[high] - sorted[low]) * (rank - low as f64)
+            })
+            .sum::<f64>()
+            / per_group[&key].len() as f64;
+        assert!(
+            (value - exact).abs() <= 0.05 * exact,
+            "{labels:?}: {value} vs exact {exact}"
+        );
+    }
+}
+
+/// Without the declaration nothing proves `data`'s samples non-negative, so
+/// the merged panes' relative bound does not carry through `avg`: P5-m1 is
+/// rejected, and with no raw samples no plan is valid.
+#[tokio::test]
+async fn an_undeclared_gauges_merged_ddsketch_panes_get_no_relative_bound_through_avg() {
+    let error = plan_p99_avg(false).await.expect_err("no valid plan");
+    assert!(error.contains("P5-m1"), "{error}");
+}