fix(optimizer): reuse exact results, grouped aggregates and maintained summaries

crates/integration-tests/tests/identical_exact_queries.rs

@@ -0,0 +1,77 @@
+//! Original issue workloads against the production candidate-connection planner.
+use asap_devtools::planner_fuzz::{read_seed_file, FuzzSettings, FuzzWorkload, Planner};
+use asap_devtools::planning_stages::run_search_over;
+use asap_plan_selection::{PlanSelectionSettings, PlanningModels, Stage1ToStage3Search};
+use asap_planner::PlannerExecutorCompileCheck;
+use serde_json::json;
+use std::path::Path;
+
+fn workload() -> FuzzWorkload {
+    let mut w = read_seed_file(
+        &Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/sharing_issue_fixtures/922.json"),
+    )
+    .unwrap();
+    w.tables = vec![json!({"name":"metrics", "time_index":0, "columns":[
+        {"name":"ts","type":"timestamp","nullable":false},
+        {"name":"city","type":"utf8","nullable":false},
+        {"name":"host","type":"utf8","nullable":false},
+        {"name":"value","type":"float64","nullable":false}
+    ]})];
+    w
+}
+
+fn cost(w: &FuzzWorkload) -> f64 {
+    // Each regression has a wall-clock planning budget. Concurrent solver
+    // runs in this test binary must not consume one another's CPU budget.
+    static PLANNING: std::sync::Mutex<()> = std::sync::Mutex::new(());
+    let _planning = PLANNING.lock().unwrap();
+    let roots = Planner::new(FuzzSettings::default()).lower(w).unwrap();
+    let planning = w.planning_workload();
+    let entries: Vec<_> = planning.query_workload.entries().collect();
+    let settings = PlanSelectionSettings::default();
+    let deployment = w.deployment();
+    let models = PlanningModels::builtin()
+        .with_plan_selection(&settings)
+        .with_deployment_capabilities(&deployment)
+        .with_executor_compile_check(&PlannerExecutorCompileCheck);
+    let run = run_search_over(
+        roots,
+        &entries,
+        &w.data,
+        models,
+        Stage1ToStage3Search {
+            milp_time_limit_ms: Some(30_000),
+            ..Default::default()
+        },
+    )
+    .unwrap();
+    let selection = run.stage3_plan_selection.as_ref().unwrap();
+    let value = selection.costs[&selection.selected]
+        .cost_per_second
+        .unwrap();
+    println!("cost={value}; selected={}", selection.selected);
+    value
+}
+
+/// #922: two identical exact percentile queries pay for their result once.
+#[test]
+fn identical_exact_percentiles_share_work() {
+    let related = workload();
+    let mut base = related.clone();
+    base.queries.truncate(1);
+    let base_cost = cost(&base);
+    let related_cost = cost(&related);
+    assert!(
+        related_cost <= base_cost * 1.001,
+        "{related_cost} vs {base_cost}"
+    );
+}
+
+/// Different ranks share the source window but still compute their own values.
+#[test]
+fn different_exact_percentiles_share_the_input_window() {
+    let mut related = workload();
+    related.queries[1].entry.query.0 = related.queries[1].entry.query.0.replace("0.99", "0.5");
+    let related_cost = cost(&related);
+    assert!(related_cost <= 0.0883, "{related_cost}");
+}

crates/integration-tests/tests/per_computation_candidates_sql.rs

@@ -222,11 +222,13 @@ async fn heap_updates_over(fields: Vec<Field>, sql: &str) -> Vec<(String, String
             }
             other => panic!("one item column: {other:?}"),
         },
-        other => panic!("a named column: {other:?}"),
+        SummaryInputExpr::Constant(value) => value.to_string(),
+        other => panic!("a named column or constant: {other:?}"),
     };
     let mut updates = Vec::new();
     for (t, target) in inventory.targets.iter().enumerate() {
-        for (c, candidate) in target.choices.candidates.iter().enumerate() {
+        for c in 0..target.choice_count() {
+            let (candidate, _) = target.computation_for(c).unwrap();
             if !matches!(
                 candidate,
                 ComputationCandidate::Summary(
@@ -261,9 +263,9 @@ const COUNT_RANKED_TOPK: &str =
     "SELECT KEY, count(*) FROM flows GROUP BY KEY ORDER BY count(*) DESC LIMIT 10";
 
 /// An Int64 key is not mistaken for the ranked value: both heap sketches
-/// compose, keyed by the key and weighted by the count.
+/// compose over raw rows, keyed by the key with unit weight.
 #[tokio::test]
-async fn topk_heap_over_an_int64_key_is_weighted_by_the_count() {
+async fn topk_heap_over_an_int64_key_counts_raw_rows() {
     let updates = heap_updates(
         "port",
         DataType::Int64,
@@ -274,15 +276,15 @@ async fn topk_heap_over_an_int64_key_is_weighted_by_the_count() {
     assert!(
         updates
             .iter()
-            .all(|(item, weight)| item == "port" && weight.starts_with("count(")),
+            .all(|(item, weight)| item == "port" && weight == "1"),
         "{updates:?}"
     );
 }
 
 /// A key named `value` is not the ranked value: the heap sketches are
-/// weighted by the count, not by the key.
+/// fed raw rows with unit weight.
 #[tokio::test]
-async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() {
+async fn topk_heap_over_a_key_named_value_counts_raw_rows() {
     let updates = heap_updates(
         "value",
         DataType::Utf8,
@@ -293,16 +295,15 @@ async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() {
     assert!(
         updates
             .iter()
-            .all(|(item, weight)| item == "value" && weight.starts_with("count(")),
+            .all(|(item, weight)| item == "value" && weight == "1"),
         "{updates:?}"
     );
 }
 
 /// SQL's per-group top-k idiom (`ROW_NUMBER() OVER (PARTITION BY city
 /// ORDER BY COUNT(*) DESC) … WHERE rn <= 5`) is a heavy-hitter top-k, as
 /// PromQL `topk by (city)` is: both heap sketches compose, keyed by the
-/// host and weighted by the count (the aggregate's one value column, so
-/// read as its sample value).
+/// host and fed raw rows with unit weight.
 #[tokio::test]
 async fn rownumber_per_group_topk_offers_heap_sketches() {
     let updates = heap_updates_over(
@@ -319,7 +320,7 @@ async fn rownumber_per_group_topk_offers_heap_sketches() {
     assert!(
         updates
             .iter()
-            .all(|(item, weight)| item == "host" && weight == "<sample value>"),
+            .all(|(item, weight)| item == "host" && weight == "1"),
         "{updates:?}"
     );
 }

crates/integration-tests/tests/sharing_issue_fixtures/886.json

@@ -0,0 +1,39 @@
+{
+  "query_workload": {
+    "language": {
+      "sql": "datafusion_sql"
+    },
+    "query_batch": null,
+    "repeating_queries": [
+      {
+        "query": "SELECT host, COUNT(*) AS c FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE GROUP BY host ORDER BY c DESC LIMIT 5",
+        "demand": {
+          "fixed_interval": 10000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.01,
+            "failure_probability": 0.001
+          }
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      }
+    ]
+  },
+  "data_workload": {
+    "arrival": {
+      "ingesting": {
+        "rows_per_second": 666.67
+      }
+    },
+    "sources": {
+      "metrics": {
+        "series_count": 100
+      }
+    }
+  }
+}

crates/integration-tests/tests/sharing_issue_fixtures/902.json

@@ -0,0 +1,73 @@
+{
+  "query_workload": {
+    "language": {
+      "sql": "datafusion_sql"
+    },
+    "query_batch": null,
+    "repeating_queries": [
+      {
+        "query": "SELECT percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE",
+        "demand": {
+          "fixed_interval": 10000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.01,
+            "failure_probability": 0.001
+          }
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      },
+      {
+        "query": "SELECT percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE AND city = 'NYC'",
+        "demand": {
+          "fixed_interval": 10000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.01,
+            "failure_probability": 0.001
+          }
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      },
+      {
+        "query": "SELECT city, percentile_cont(0.9) WITHIN GROUP (ORDER BY value) FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE GROUP BY city",
+        "demand": {
+          "fixed_interval": 10000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.01,
+            "failure_probability": 0.001
+          }
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      }
+    ]
+  },
+  "data_workload": {
+    "arrival": {
+      "ingesting": {
+        "rows_per_second": 666.67
+      }
+    },
+    "sources": {
+      "metrics": {
+        "series_count": 100
+      }
+    }
+  }
+}

crates/integration-tests/tests/sharing_issue_fixtures/922.json

@@ -0,0 +1,69 @@
+{
+  "query_workload": {
+    "language": {
+      "sql": "datafusion_sql"
+    },
+    "query_batch": null,
+    "repeating_queries": [
+      {
+        "query": "SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value) AS v FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE",
+        "demand": {
+          "fixed_interval": 30000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.0,
+            "failure_probability": 0.0,
+            "accuracy_scope": "whole_result"
+          },
+          "max_latency_ms": null
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      },
+      {
+        "query": "SELECT percentile_cont(0.99) WITHIN GROUP (ORDER BY value) AS v FROM metrics WHERE ts > now() - INTERVAL '5' MINUTE",
+        "demand": {
+          "fixed_interval": 30000
+        },
+        "requirements": {
+          "accuracy": {
+            "max_error": 0.0,
+            "failure_probability": 0.0,
+            "accuracy_scope": "whole_result"
+          },
+          "max_latency_ms": null
+        },
+        "predictability": {
+          "predictable": {
+            "known_at": null
+          }
+        }
+      }
+    ]
+  },
+  "data_workload": {
+    "arrival": {
+      "ingesting": {
+        "rows_per_second": 1000.0
+      }
+    },
+    "sources": {
+      "metrics": {
+        "series_count": 10000,
+        "metric_type": null,
+        "values_non_negative": true,
+        "sample_interval": 10000,
+        "stream_norms": null,
+        "max_distinct_per_evaluation": null,
+        "distinct_values_per_label": {
+          "city": 10,
+          "host": 100
+        }
+      }
+    }
+  }
+}