fix: select only compilable populations; tighten readout decoding

Review follow-ups: a population is selected only if Planner compiles its
readout, so a failed compile runs the query exactly instead of failing the
deployment. Grouped population outputs must carry exactly the grouping
labels. An Int64 count is the sample only when it is the sole numeric
column. The summation difference from Prometheus is documented.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: zzylol

control_plane/src/physical/workload_cost.rs

@@ -848,7 +848,12 @@ fn enumerate_frontier_candidates(
         .map(|root| {
             root.as_ref()
                 .and_then(|root| strategy.candidate(root))
-                .filter(|node| super::maintained_population::supported_node(node))
+                .filter(|node| {
+                    // Every readout must compile to the Planner program installed with it.
+                    super::maintained_population::supported_node(node)
+                        && asap_physical_operators::physical_planner::promql_rows::compile_current_series_readout(node)
+                            .is_ok()
+                })
         })
         .collect();
     if maintained_roots.iter().any(Option::is_some) {
@@ -1044,6 +1049,37 @@ mod tests {
             })));
     }
 
+    /// Planner cannot yet read out `without` groups, so such a population is
+    /// never selected, while the `by` form is.
+    #[test]
+    fn without_grouping_selects_no_population() {
+        for (query, supported) in [
+            ("quantile without (pod) (0.5, m)", false),
+            ("quantile by (job) (0.5, m)", true),
+        ] {
+            let root = std::rc::Rc::new(
+                asap_physical_operators::physical_planner::promql_rows::with_series_identity(
+                    &crate::query_parser::parse_query_expr_canonical(
+                        query,
+                        crate::types::AccuracyTarget::Exact,
+                    )
+                    .unwrap(),
+                )
+                .unwrap(),
+            );
+            let strategy =
+                asap_aware_mapping::maintained_population::MaintainedPopulationStrategy::new(
+                    std::slice::from_ref(&root),
+                );
+            let candidate = strategy.candidate(&root).unwrap();
+            assert_eq!(
+                super::super::maintained_population::supported_node(&candidate),
+                supported,
+                "{query}"
+            );
+        }
+    }
+
     /// Instant counts select current membership, never accumulated observations.
     #[test]
     fn local_grouped_count_has_a_bindable_candidate() {

crates/asap_types/src/query_plan/native.rs

@@ -54,12 +54,19 @@ impl QueryPlanEntry {
                 SummaryFamilyType::Plain(DataType::Float64 | DataType::Int64)
             )
         };
+        let labels = output
+            .schema
+            .fields
+            .iter()
+            .filter(|field| !numeric(field))
+            .map(|field| {
+                (field.dtype == SummaryFamilyType::Plain(DataType::Utf8)).then_some(&field.name)
+            })
+            .collect::<Option<BTreeSet<_>>>();
         let grouped_value = output.schema.time_index.is_none()
             && output.schema.fields.iter().filter(|f| numeric(f)).count() == 1
-            && output.schema.fields.iter().all(|field| {
-                numeric(field)
-                    || (field.dtype == SummaryFamilyType::Plain(DataType::Utf8)
-                        && population.grouping.labels.contains(&field.name))
+            && labels.is_some_and(|labels| {
+                labels == population.grouping.labels.iter().collect::<BTreeSet<_>>()
             });
         if input.schema != output.schema && !grouped_value {
             return Err(invalid(

data_plane/src/query_engines/asap_query_engine/logical_dag/native_values.rs

@@ -508,6 +508,43 @@ mod tests {
         );
     }
 
+    // A population count is an Int64 sample only when it is the sole numeric
+    // column; a Float64 value always wins over an Int64 helper column.
+    #[test]
+    fn selected_program_reads_counts_and_prefers_float_values() {
+        for (fields, row, expected) in [
+            (
+                vec![("job", DataType::Utf8), ("count", DataType::Int64)],
+                vec![Value::Utf8("api".into()), Value::Int64(3)],
+                3.,
+            ),
+            (
+                vec![("members", DataType::Int64), ("value", DataType::Float64)],
+                vec![Value::Int64(7), Value::Float64(2.5)],
+                2.5,
+            ),
+        ] {
+            let input = schema(&fields);
+            let plan = CompiledPhysicalDag::from_operators(
+                [(0, InputContract::bounded(input.clone()))].into(),
+                Default::default(),
+                vec![0],
+            )
+            .unwrap();
+            let (result, _) = execute_batches(&plan, 1 << 20, 1, 42, false, |_, contract| {
+                Ok(BoundInput::Rows(Batch::try_new(
+                    contract.schema.clone(),
+                    vec![row.clone()],
+                )?))
+            })
+            .unwrap();
+            let crate::query_engines::query_result::QueryResult::Vector(result) = result else {
+                panic!("vector expected")
+            };
+            assert_eq!(result.values[0].value, expected);
+        }
+    }
+
     // Count-like values are bound as integers only when the protocol sample is exact.
     #[test]
     fn integer_input_binding_preserves_type_and_rejects_rounding() {
@@ -867,17 +904,25 @@ fn execute_batches(
                 .fields
                 .iter()
                 .position(|field| field.name == SERIES_IDENTITY_COLUMN);
-            let value = batch
-                .schema()
-                .fields
-                .iter()
-                .position(|field| {
-                    matches!(
-                        field.dtype,
-                        SummaryFamilyType::Plain(DataType::Float64 | DataType::Int64)
-                    )
-                })
-                .ok_or_else(|| miss("physical output loses sample value"))?;
+            let fields = &batch.schema().fields;
+            let typed = |dtype: DataType| {
+                fields
+                    .iter()
+                    .enumerate()
+                    .filter(|(_, field)| field.dtype == SummaryFamilyType::Plain(dtype.clone()))
+                    .map(|(i, _)| i)
+                    .collect::<Vec<_>>()
+            };
+            // The sample is the Float64 column; an Int64 count is the sample
+            // only when it is the sole numeric column.
+            let value = match (
+                typed(DataType::Float64).as_slice(),
+                typed(DataType::Int64).as_slice(),
+            ) {
+                ([value, ..], _) => *value,
+                ([], [count]) => *count,
+                _ => return Err(miss("physical output loses sample value")),
+            };
             for row in batch.rows() {
                 let sample = &match row[value] {
                     Value::Float64(sample) => sample,

data_plane/src/storage_engines/sketch_db/current_series.rs

@@ -156,7 +156,7 @@ impl Population {
             .map(|(k, v)| (k.clone(), v.clone()))
             .collect();
         self.remove(&labels);
-        // Bound the retained keys, tree nodes, and shared readout caches together.
+        // Bound the retained keys and tree nodes together.
         let bytes = 1024
             + labels
                 .iter()

docs/developer_docs/query-engine/current-series-aggregations.md

@@ -1,4 +1,4 @@
-# Shared current-series quantiles and TopK
+# Shared current-series aggregations
 
 Backend-local PromQL workload compilation can export a maintained current-value
 alternative for `quantile(q, metric)`, `topk(k, metric)`, `sum`, `count` and
@@ -50,6 +50,9 @@ A native exact alternative remains available for cost selection and execution fa
   unobserved generation or exceeded resource bounds cause native fallback.
 - State is in memory. Restart and generation replacement require warmup again;
   historical range queries continue to use native execution.
+- Sum and average use Planner's aggregate operator, which adds values in order
+  without Prometheus' compensated summation; results can differ in the last
+  bits, and an average whose sum overflows is infinite.
 - Populations divide the configured retained-summary memory budget and cap series
   cardinality. Bounds include conservative space for labels and trees.
   The existing Remote Write adapter accepts finite sample values and stale markers.