SQL per-group top-k (ROW_NUMBER … rn <= k) plans as PromQL topk by; exact plans compile; compile check covers stored-output builds

crates/devtools/tests/cross_language.rs

@@ -162,24 +162,78 @@ async fn ascending_count_ranked_topk_stays_generic_in_both_languages() {
     );
 }
 
-/// Row-number filters retain their computed column and outer projection scope.
+/// The partitioned heavy-hitter of `qe`: its `k` and the names of the
+/// columns it partitions by, if `qe` is an outer `Aggregate([TopK{k}])`.
+fn partitioned_top_k(qe: &OperatorNode) -> Option<(usize, Vec<String>)> {
+    let Some(NonASAPOp::Aggregate {
+        reduction,
+        measures,
+        child,
+        ..
+    }) = qe.non_asap()
+    else {
+        return None;
+    };
+    let [AggIntent::TopK { k, .. }] = measures.as_slice() else {
+        return None;
+    };
+    let partition = reduction
+        .expect_reduce()
+        .iter()
+        .map(|&column| child.schema.fields[column].name.clone())
+        .collect();
+    Some((*k, partition))
+}
 
+/// SQL's per-partition top-k idiom (`ROW_NUMBER() OVER (PARTITION BY
+/// region ORDER BY COUNT(*) DESC) … WHERE rn <= 5`) is the heavy-hitter
+/// PromQL `topk by (region)` lowers to: the top 5 per region by count, an
+/// outer `TopK{5}` grouped by `region` over the explicit inner Count.
 #[tokio::test]
-async fn sql_rownumber_count_preserves_window_schema() {
-    let query=sql("SELECT service, region, v FROM (SELECT service, region, COUNT(*) AS v, ROW_NUMBER() OVER (PARTITION BY region ORDER BY COUNT(*) DESC) AS rn FROM metrics GROUP BY service, region) t WHERE rn <= 5").await;
-    query.validate_structure().unwrap();
-    assert_eq!(query.schema.fields.len(), 3);
-    assert!(OperatorNode::reachable(&query)
+async fn sql_rownumber_count_topk_is_promqls_topk_by() {
+    let s = sql(
+        "SELECT service, region, v FROM (SELECT service, region, COUNT(*) AS v, \
+         ROW_NUMBER() OVER (PARTITION BY region ORDER BY COUNT(*) DESC) AS rn \
+         FROM metrics GROUP BY service, region) t WHERE rn <= 5",
+    )
+    .await;
+    let p = promql(
+        "topk by (region) (5, sum by (service, region) (count_over_time(http_requests_total[5m])))",
+    );
+    s.validate_structure().unwrap();
+    let region = vec!["region".to_string()];
+    assert_eq!(partitioned_top_k(&s), Some((5, region.clone())), "{s:?}");
+    assert_eq!(partitioned_top_k(&p), Some((5, region)), "{p:?}");
+    assert!(heavy_hitter(&s).is_some(), "over the explicit inner Count");
+    assert!(!OperatorNode::reachable(&s)
         .iter()
         .any(|node| matches!(node.non_asap(), Some(NonASAPOp::SQLWindowFunc { .. }))));
 }
 
+/// An avg ranking is no heavy-hitter: the generic per-partition
+/// `Limit{5}{Sort}` that PromQL `topk by (region)` over an average lowers
+/// to, with the query's three columns.
 #[tokio::test]
-async fn sql_rownumber_avg_preserves_window_schema() {
-    let query=sql("SELECT service, region, v FROM (SELECT service, region, AVG(latency) AS v, ROW_NUMBER() OVER (PARTITION BY region ORDER BY AVG(latency) DESC) AS rn FROM metrics GROUP BY service, region) t WHERE rn <= 5").await;
+async fn sql_rownumber_avg_topk_is_a_partitioned_sort_limit() {
+    let query = sql(
+        "SELECT service, region, v FROM (SELECT service, region, AVG(latency) AS v, \
+         ROW_NUMBER() OVER (PARTITION BY region ORDER BY AVG(latency) DESC) AS rn \
+         FROM metrics GROUP BY service, region) t WHERE rn <= 5",
+    )
+    .await;
     query.validate_structure().unwrap();
     assert_eq!(query.schema.fields.len(), 3);
-    assert!(OperatorNode::reachable(&query)
+    let Some(NonASAPOp::Limit {
+        n: Some(5),
+        partition_by,
+        child,
+        ..
+    }) = query.non_asap()
+    else {
+        panic!("a per-partition Limit: {query:?}");
+    };
+    assert_eq!(child.schema.fields[partition_by[0]].name, "region");
+    assert!(!OperatorNode::reachable(&query)
         .iter()
         .any(|node| matches!(node.non_asap(), Some(NonASAPOp::SQLWindowFunc { .. }))));
 }
@@ -196,3 +250,25 @@ async fn offset_defeats_heavy_hitter_promotion() {
         "OFFSET must not promote to TopK: {s:?}"
     );
 }
+
+/// Heavy-hitter promotion changes a node's columns, so it applies only to
+/// a ranking nothing reads by position. `topk(5, sum by (job) (m))` at the
+/// root is promoted; under `count(...)`, which reads its columns, it stays
+/// the order-by-value `Limit{Sort}`.
+#[tokio::test]
+async fn a_promql_ranking_read_by_position_stays_a_sort_limit() {
+    let root = promql("topk(5, sum by (job) (http_requests_total))");
+    assert!(
+        matches!(root.non_asap(), Some(NonASAPOp::Aggregate { measures, .. })
+            if matches!(measures.as_slice(), [AggIntent::TopK { .. }])),
+        "{root:?}"
+    );
+    let nested = promql("count(topk(5, sum by (job) (http_requests_total)))");
+    let Some(NonASAPOp::Aggregate { child, .. }) = nested.non_asap() else {
+        panic!("count at the root: {nested:?}");
+    };
+    assert!(
+        matches!(child.non_asap(), Some(NonASAPOp::Limit { .. })),
+        "{child:?}"
+    );
+}

crates/devtools/tests/stage_pipeline.rs

@@ -1895,7 +1895,10 @@ fn mixed_language_workload_files(sql_arrival: Value, sql_facts: Value) -> (PathB
     (promql_path, sql_path)
 }
 
-const FLOWS_TABLE: &str = r#"{"name": "flows", "columns": [{"name": "ts", "type": "timestamp"}, {"name": "service", "type": "utf8", "nullable": false}], "time_index": 0}"#;
+/// `flows(ts, service)`. The time index is declared non-null: the executor
+/// builds no count over a nullable time column, so a plan of the SQL query
+/// over it would not compile.
+const FLOWS_TABLE: &str = r#"{"name": "flows", "columns": [{"name": "ts", "type": "timestamp", "nullable": false}, {"name": "service", "type": "utf8", "nullable": false}], "time_index": 0}"#;
 
 /// One run can plan PromQL and SQL queries together: `--workload` repeats,
 /// each file is lowered by its own language's frontend (SQL over the

crates/integration-tests/tests/per_computation_candidates_sql.rs

@@ -184,19 +184,27 @@ async fn a_sql_fold_over_estimates_leaves_its_row_count_unbound() {
 /// candidate of `sql` over `flows(key: key_type, bytes)` that picks a heap
 /// sketch for its top-k.
 async fn heap_updates(key: &str, key_type: DataType, sql: &str) -> Vec<(String, String)> {
-    let catalog = SqlCatalog::new().with_table(
-        "flows",
-        Schema::new(vec![
+    heap_updates_over(
+        vec![
             Field::plain(key, key_type, false),
             Field::plain("bytes", DataType::Float64, false),
-        ]),
-    );
+        ],
+        sql,
+    )
+    .await
+}
+
+/// [`heap_updates`] over `flows(fields)`.
+async fn heap_updates_over(fields: Vec<Field>, sql: &str) -> Vec<(String, String)> {
+    let catalog = SqlCatalog::new().with_table("flows", Schema::new(fields));
     let accuracy = AccuracyTarget::max_error(0.01).failure_probability(0.01);
     let root = lower_sql(sql, &catalog, accuracy).await.unwrap();
     let inventory =
         generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap();
     let name = |expr: &SummaryInputExpr| match expr {
         SummaryInputExpr::Column(ColumnRef::Named(name)) => name.clone(),
+        // The ranked value when it is the child's one sample value.
+        SummaryInputExpr::Column(ColumnRef::SampleValue) => "<sample value>".into(),
         SummaryInputExpr::Tuple(items) => match items.as_slice() {
             [SummaryInputExpr::Column(ColumnRef::Named(name))] => name.clone(),
             other => panic!("one item column: {other:?}"),
@@ -276,3 +284,29 @@ async fn topk_heap_over_a_key_named_value_is_weighted_by_the_count() {
         "{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).
+#[tokio::test]
+async fn rownumber_per_group_topk_offers_heap_sketches() {
+    let updates = heap_updates_over(
+        vec![
+            Field::plain("city", DataType::Utf8, false),
+            Field::plain("host", DataType::Utf8, false),
+        ],
+        "SELECT city, host, c FROM (SELECT city, host, COUNT(*) AS c, ROW_NUMBER() OVER \
+         (PARTITION BY city ORDER BY COUNT(*) DESC) AS rn FROM flows GROUP BY city, host) \
+         WHERE rn <= 5",
+    )
+    .await;
+    assert_eq!(updates.len(), 2, "{updates:?}");
+    assert!(
+        updates
+            .iter()
+            .all(|(item, weight)| item == "host" && weight == "<sample value>"),
+        "{updates:?}"
+    );
+}

crates/integration-tests/tests/stage2_physical_asap_optimization_sql.rs

@@ -1,7 +1,8 @@
 //! SQL heavy-hitter queries through Stage 1 (every Pass 1 choice composed)
 //! and physical ASAP-aware optimization: exact top-k sorts by the ranked
-//! measure, descending, whatever the group keys are called or typed, and no
-//! valid candidate is pruned.
+//! measure, descending, whatever the group keys are called or typed, no
+//! valid candidate is pruned, and every candidate answers the ranked value
+//! then the group keys naming each ranked item.
 
 use asap_frontend_sql::{lower_sql, SqlCatalog};
 use asap_logical_asap_optimization::per_computation_candidates::{
@@ -155,7 +156,7 @@ async fn group_key_named_value_ranks_by_count() {
     check(
         "SELECT value, COUNT(*) AS c FROM metrics GROUP BY value ORDER BY c DESC LIMIT 5",
         "count(Int64(1))",
-        &["topk_5"],
+        &["topk_5", "value"],
     )
     .await;
 }
@@ -166,7 +167,7 @@ async fn integer_group_key_ranks_by_count() {
     check(
         "SELECT port, COUNT(*) AS c FROM metrics GROUP BY port ORDER BY c DESC LIMIT 5",
         "count(Int64(1))",
-        &["topk_5"],
+        &["topk_5", "port"],
     )
     .await;
 }
@@ -178,7 +179,7 @@ async fn multi_key_ranks_by_count() {
         "SELECT host, port, COUNT(*) AS c FROM metrics GROUP BY host, port \
          ORDER BY c DESC LIMIT 5",
         "count(Int64(1))",
-        &["topk_5"],
+        &["topk_5", "host", "port"],
     )
     .await;
 }
@@ -189,7 +190,7 @@ async fn sum_ranking_ranks_by_sum() {
     check(
         "SELECT port, SUM(value) AS s FROM metrics GROUP BY port ORDER BY s DESC LIMIT 5",
         "sum(metrics.value)",
-        &["topk_5"],
+        &["topk_5", "port"],
     )
     .await;
 }

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

@@ -2982,11 +2982,10 @@ fn whole_expression(
 
 /// A heap's top-k `estimate` (`[partitions.., items.., topk]`, one row per
 /// ranked item) in `target`'s columns: its `[keys.., topk_k]`, with
-/// `topk_k` the estimated Float64 weight, then, over PromQL rows, the item
-/// columns. This is the exact top-k's physical shape (its ranked rows'
-/// value, then their other labels; a SQL result keeps only its logical
-/// columns), so readers of `[keys.., topk_k]` find those columns where they
-/// were.
+/// `topk_k` the estimated Float64 weight, then the item columns. This is
+/// the exact top-k's physical shape (its ranked rows' value, then the
+/// columns naming each ranked item: PromQL labels, SQL group keys), so
+/// readers of `[keys.., topk_k]` find those columns where they were.
 fn ranked_as_topk(
     target: &OperatorNode,
     estimate: Rc<OperatorNode>,
@@ -3014,18 +3013,14 @@ fn ranked_as_topk(
                 ))
         })
         .collect::<Result<Vec<_>, _>>()?;
-    let promql = match target.non_asap() {
-        Some(NonASAPOp::Aggregate { child, .. }) => reads_only_time_series(child),
-        _ => false,
-    };
     let cols = key_columns
         .iter()
         .zip(keys)
         .map(|(&index, key)| column(index, &key.name))
         .chain(std::iter::once(column(estimated, &ranked.name)))
         .chain(
             (0..estimated)
-                .filter(|index| promql && !key_columns.contains(index))
+                .filter(|index| !key_columns.contains(index))
                 .map(|index| column(index, &fields[index].name)),
         )
         .collect();

crates/physical-asap-optimization/src/lib.rs

@@ -87,7 +87,9 @@
 //! candidate's and as the frontend declares. Over PromQL rows it also keeps
 //! the series identity, after them: `topk` answers whole input series, so
 //! each answer carries the chosen series' full label set. PromQL rows without
-//! one (e.g. of `sum by (job)`) keep their other string label columns instead.
+//! one (e.g. of `sum by (job)`) keep their other string label columns instead,
+//! and SQL rows their other group keys (the ranked item, e.g. `host` of
+//! `GROUP BY host ORDER BY COUNT(*) DESC LIMIT k`).
 //! `topk_k` carries the ranked value's type and nullability (the frontend
 //! declares a non-null `Utf8` placeholder); casting the value to `Utf8` would
 //! not make it the same answer.
@@ -1064,14 +1066,17 @@ fn implement_topk(
             alias: Some(asap_types::ir::schema::PROMQL_SERIES_IDENTITY.into()),
             expr: ScalarExpr::Column(identity),
         });
-    } else if reads_only_time_series(&child) {
-        // PromQL rows without it (e.g. those of `sum by (job)`) carry their
-        // labels as plain string columns: each answer keeps the ones the
-        // logical columns do not, so `topk(k, sum by (job) (...))` answers
-        // each job. A SQL result keeps only its logical columns.
+    } else {
+        // Otherwise each answer keeps the columns naming its ranked item
+        // that the logical columns do not: over PromQL rows without the
+        // identity (e.g. those of `sum by (job)`) their string labels, so
+        // `topk(k, sum by (job) (...))` answers each job; over a SQL
+        // aggregate its other group keys, so `ORDER BY COUNT(*) DESC LIMIT
+        // k` over `GROUP BY host` answers each host.
+        let promql = reads_only_time_series(&child);
         for (index, field) in child.schema.fields.iter().enumerate() {
-            let is_label = field.dtype == FieldDataType::Plain(DataType::Utf8);
-            if is_label && index != value && !key_fields.iter().any(|k| k.name == field.name) {
+            let names_item = !promql || field.dtype == FieldDataType::Plain(DataType::Utf8);
+            if names_item && index != value && !key_fields.iter().any(|k| k.name == field.name) {
                 cols.push(ProjectItem {
                     alias: Some(field.name.clone()),
                     expr: ScalarExpr::Column(index),

crates/planner/src/installable.rs

@@ -435,6 +435,13 @@ impl ExecutorCompileCheck for PlannerExecutorCompileCheck {
         };
         match installable() {
             Ok(plan) => plan.check_compiles(),
+            // Deriving the plan's stored outputs already compiled each
+            // build, and the executor refused one: the plan does not
+            // compile, whichever run builds that output, as `check_compiles`
+            // reports a build it refuses.
+            Err(InstallablePlanError::StoredOutputNotExecutable { node_id, reason }) => {
+                Err(format!("stored output {node_id}: {reason}"))
+            }
             // Not installable for another reason, which `plan_installable`
             // reports as such (e.g. a scalar root); not the compiler's to say.
             Err(_) => Ok(()),