fix(planner): only a Filter over the aggregate is its HAVING

A Filter that reads an aggregate only from a scalar expression (PromQL
scalar(), subqueries) stays a plain consumer instead of a HavingEntry;
filters above are deduplicated by pointer. Print the InvalidAccuracy
root once, add tests for Project/scalar consumers, mixed SQL HAVING,
without/PerEntity and annotated non-root nodes, and rewrap the doc.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW

Author: zzylol

crates/asap-aware-mapping/src/logical_candidates.rs

@@ -141,10 +141,12 @@ pub enum LogicalCandidateError {
     AssignedPhysicalAnnotation,
     #[error("logical candidate input already contains ASAP operators")]
     ContainsAsap,
-    /// `root` and `measure` locate the target; [`local_candidates_for_intent`]
-    /// alone leaves them `None`.
+    /// `root` (the root id's `Debug` form) and `measure` locate the target;
+    /// [`local_candidates_for_intent`] alone leaves them `None`.
     #[error(
-        "approximate accuracy of root {root:?}, measure {measure:?} requires epsilon and delta in (0, 1)"
+        "approximate accuracy of root {}, measure {} requires epsilon and delta in (0, 1)",
+        root.as_deref().unwrap_or("?"),
+        measure.map_or_else(|| "?".to_string(), |measure| measure.to_string())
     )]
     InvalidAccuracy {
         root: Option<String>,
@@ -312,14 +314,20 @@ pub fn enumerate_local_logical_candidates<Id: std::fmt::Debug>(
             })
             .collect();
         let parents = &consumers[&Rc::as_ptr(&node)];
-        let filters_above: Vec<(&Rc<OperatorNode>, &Predicate)> = parents
-            .iter()
-            .flatten()
-            .filter_map(|parent| match parent.non_asap() {
-                Some(NonASAPOp::Filter { pred, .. }) => Some((parent, pred)),
-                _ => None,
-            })
-            .collect();
+        // Only a `Filter` whose input is this aggregate is its `HAVING`; one
+        // that reads it from a scalar expression stays a plain consumer.
+        let mut filters_above: Vec<(&Rc<OperatorNode>, &Predicate)> = Vec::new();
+        for parent in parents.iter().flatten() {
+            if let Some(NonASAPOp::Filter { child, pred }) = parent.non_asap() {
+                if Rc::ptr_eq(child, &node)
+                    && !filters_above
+                        .iter()
+                        .any(|(filter, _)| Rc::ptr_eq(filter, parent))
+                {
+                    filters_above.push((parent, pred));
+                }
+            }
+        }
         for (filter, predicate) in &filters_above {
             let shared_by_all = filters_above.len() == parents.len()
                 && filters_above.iter().all(|(_, other)| other == predicate);

crates/asap-aware-mapping/tests/logical_candidates.rs

@@ -254,7 +254,10 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() {
     .clone();
     producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime);
     assert!(matches!(
-        enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]),
+        enumerate_local_logical_candidates(vec![(
+            0,
+            QueryRoot::Operator(Rc::new(producer.clone()))
+        )]),
         Err(LogicalCandidateError::AssignedPhysicalAnnotation)
     ));
     let mut stored = (*aggregate(AggIntent::Count {
@@ -264,9 +267,17 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() {
     stored.materialization =
         asap_types::post_asap::Materialization::MatAtQuery(asap_types::post_asap::Medium::Memory);
     assert!(matches!(
-        enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored)))]),
+        enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored.clone())))]),
         Err(LogicalCandidateError::AssignedPhysicalAnnotation)
     ));
+    // An annotated node below an unannotated root is rejected too.
+    for annotated in [producer, stored] {
+        let root = having_filter(Rc::new(annotated), always());
+        assert!(matches!(
+            enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]),
+            Err(LogicalCandidateError::AssignedPhysicalAnnotation)
+        ));
+    }
     let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg {
         child: scan(false),
         family: FieldDataType::Sketch(
@@ -331,6 +342,10 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() {
         ),
         "{error:?}"
     );
+    assert!(
+        error.to_string().contains("root \"q\", measure 1 "),
+        "{error}"
+    );
 }
 
 /// Local TopK keeps both declared heap substrates without choosing an implementation.
@@ -674,3 +689,153 @@ fn example_1_exact_q2_prunes_topk_heaps() {
         && choices.pruned.is_empty()));
     assert_eq!(found.len(), 2, "{found:#?}");
 }
+
+/// The `HAVING` entries of the target whose measure satisfies `is`.
+fn having_of(
+    targets: &[asap_aware_mapping::logical_candidates::LocalLogicalTarget],
+    is: impl Fn(&AggIntent) -> bool,
+) -> Vec<HavingEntry> {
+    let target = targets
+        .iter()
+        .find(|target| match target.target.non_asap() {
+            Some(NonASAPOp::Aggregate { measures, .. }) => is(&measures[target.measure]),
+            _ => false,
+        })
+        .expect("target");
+    target.having.clone()
+}
+
+/// `Column(column) > ScalarSubquery(subquery)`.
+fn above_scalar(column: usize, subquery: Rc<OperatorNode>) -> Predicate {
+    Predicate(ScalarExpr::Compare {
+        left: Box::new(ScalarExpr::Column(column)),
+        op: CompareOpKind::Gt,
+        right: Box::new(ScalarExpr::ScalarSubquery(subquery)),
+        semantics: ExprSemantics::Sql,
+    })
+}
+
+/// A `Filter` that reads an aggregate only from its predicate is not that
+/// aggregate's `HAVING`; one that also filters the aggregate gives one entry.
+#[test]
+fn scalar_reference_filter_is_not_having() {
+    let count = || {
+        vec![AggIntent::Count {
+            accuracy: approximate(),
+        }]
+    };
+    let agg = aggregate_over(scan(false), count(), vec![], None);
+    let root = having_filter(scan(false), above_scalar(1, agg.clone()));
+    let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))])
+        .unwrap()
+        .targets;
+    assert_eq!(targets.len(), 1);
+    assert!(Rc::ptr_eq(&targets[0].target, &agg));
+    assert!(targets[0].having.is_empty(), "{:?}", targets[0].having);
+
+    let agg = aggregate_over(scan(false), count(), vec![], None);
+    let root = having_filter(agg.clone(), above_scalar(0, agg.clone()));
+    let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root.clone()))])
+        .unwrap()
+        .targets;
+    assert_eq!(targets.len(), 1);
+    assert_eq!(
+        targets[0].having,
+        vec![HavingEntry {
+            filter: Some(root),
+            predicate: above_scalar(0, agg),
+            placement: HavingPlacement::AboveOutput,
+        }]
+    );
+}
+
+/// PromQL `scalar(...)` operands are scalar references, not `HAVING`.
+#[test]
+fn promql_scalar_operand_is_not_having() {
+    let roots = implicit_exact_promql("http_requests_total > scalar(sum(http_requests_total))");
+    let targets = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect())
+        .unwrap()
+        .targets;
+    assert!(!targets.is_empty());
+    for target in &targets {
+        assert!(target.having.is_empty(), "{:?}", target.having);
+    }
+
+    let roots = implicit_exact_promql("sum by (job) (x) > scalar(count(x))");
+    let targets = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect())
+        .unwrap()
+        .targets;
+    assert!(having_of(&targets, |intent| matches!(intent, AggIntent::Count { .. })).is_empty());
+    let sum = having_of(&targets, |intent| matches!(intent, AggIntent::Sum { .. }));
+    assert_eq!(sum.len(), 1, "{sum:?}");
+    assert_eq!(sum[0].placement, HavingPlacement::AboveOutput);
+}
+
+/// A non-root, non-`Filter` consumer keeps a key-only `HAVING` above.
+#[test]
+fn having_shared_with_a_project_consumer_stays_above() {
+    let node = grouped_count(None);
+    let key = equals(0, ScalarValue::Utf8("a".into()));
+    let filtered = having_filter(node.clone(), key);
+    let project = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project {
+        cols: vec![asap_types::ir::ProjectItem {
+            alias: None,
+            expr: ScalarExpr::Column(0),
+        }],
+        qualifier: None,
+        child: node,
+    }))
+    .unwrap();
+    let targets = enumerate_local_logical_candidates(vec![
+        (0, QueryRoot::Operator(filtered)),
+        (1, QueryRoot::Operator(project)),
+    ])
+    .unwrap()
+    .targets;
+    assert_eq!(targets.len(), 1);
+    let placements: Vec<_> = targets[0]
+        .having
+        .iter()
+        .map(|entry| entry.placement)
+        .collect();
+    assert_eq!(placements, vec![HavingPlacement::AboveOutput]);
+}
+
+/// `without(...)` and per-entity aggregates have no fixed key prefix, so a
+/// `HAVING` over column 0 stays above.
+#[test]
+fn without_and_per_entity_having_stays_above() {
+    for reduction in [
+        Reduction::Reduce(asap_types::ir::operator_properties::GroupKeys::without(
+            vec![1],
+        )),
+        Reduction::PerEntity,
+    ] {
+        let node = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate {
+            child: scan(false),
+            reduction: reduction.clone(),
+            measures: vec![AggIntent::Count {
+                accuracy: approximate(),
+            }],
+            output_names: vec![],
+            filters: vec![],
+            having: None,
+        }))
+        .unwrap();
+        let root = having_filter(node, equals(0, ScalarValue::Utf8("a".into())));
+        let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))])
+            .unwrap()
+            .targets;
+        assert_eq!(targets.len(), 1, "{reduction:?}");
+        let placements: Vec<_> = targets[0]
+            .having
+            .iter()
+            .map(|entry| entry.placement)
+            .collect();
+        assert_eq!(
+            placements,
+            vec![HavingPlacement::AboveOutput],
+            "{reduction:?}"
+        );
+    }
+}

crates/integration-tests/tests/logical_candidates_sql.rs

@@ -53,3 +53,11 @@ async fn global_constant_having_stays_above() {
     let targets = targets("SELECT count(*) FROM flows HAVING false").await;
     assert_eq!(placements(&targets), vec![HavingPlacement::AboveOutput]);
 }
+
+#[tokio::test]
+async fn mixed_key_and_value_having_stays_above() {
+    let targets =
+        targets("SELECT src, count(*) FROM flows GROUP BY src HAVING src = 'a' AND count(*) > 10")
+            .await;
+    assert_eq!(placements(&targets), vec![HavingPlacement::AboveOutput]);
+}

docs/develop_docs/local-logical-candidates.md

@@ -49,24 +49,28 @@ the target keeps every candidate and records how the predicate is applied:
 - `HAVING` is recorded as `LocalLogicalTarget.having`, one `HavingEntry`
   (consumer `Filter`, predicate, placement) each. Only the adjacent form is
   recorded for push-down: the aggregate's own `having` (an entry with no
-  `Filter`) and a `Filter` directly above the aggregate, which is how the SQL
-  frontend lowers `HAVING`. A predicate that reads only `by(...)` group keys
-  is `HavingPlacement::BelowAggregate`: it is pushed below the aggregate as a
-  row predicate, like a measure filter. A consumer `Filter`'s predicate is
-  pushed below only when every consumer of the aggregate applies the same
-  predicate; otherwise another consumer would lose rows. Any other `HAVING`,
-  and every `HAVING` of a global aggregate (no group keys, so its one output
-  row exists even with no input rows), is `HavingPlacement::AboveOutput`: it
-  stays a `Filter` above the finalize/estimate output, and approximate
-  candidates of the target carry `guarantee = None`, since membership near
-  the threshold may change. A value predicate anywhere above an approximate
-  output, adjacent or not, gets `guarantee = None` through guarantee
-  propagation (#575); multi-level push-down is tracked in #664.
+  `Filter`) and a `Filter` whose input is the aggregate, which is how the SQL
+  frontend lowers `HAVING`. A `Filter` that reads the aggregate only from a
+  scalar expression (for example PromQL `scalar(...)`) is not its `HAVING`,
+  though it still counts as a consumer. A predicate that reads only `by(...)`
+  group keys is `HavingPlacement::BelowAggregate`: it is pushed below the
+  aggregate as a row predicate, like a measure filter. A consumer `Filter`'s
+  predicate is pushed below only when every consumer of the aggregate applies
+  the same predicate; otherwise another consumer would lose rows. Any other
+  `HAVING`, and every `HAVING` of a global aggregate (no group keys, so its
+  one output row exists even with no input rows), is
+  `HavingPlacement::AboveOutput`: it stays a `Filter` above the
+  finalize/estimate output, and approximate candidates of the target carry
+  `guarantee = None`, since membership near the threshold may change. A value
+  predicate anywhere above an approximate output, adjacent or not, gets
+  `guarantee = None` through guarantee propagation (#575); multi-level
+  push-down is tracked in #664.
 
 For example, an approximate single-column distinct count offers exact execution,
 HLL, Theta, KMV and UnivMon; an approximate `count` offers its exact `Count`
-accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are candidate choices, not assessed accuracy
-certificates. Catalog order is stable and has no cost/preference meaning.
+accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are
+candidate choices, not assessed accuracy certificates. Catalog order is stable
+and has no cost/preference meaning.
 
 ## Pruning
 
@@ -97,9 +101,8 @@ sub-DAG, including applying measure filters and `HAVING` as recorded on the
 target, and validating its structure is still Pass 1 work, done later when
 candidates are realized (#575); Pass 2 then retains independent and shared
 alternatives before physical planning and complete workload selection. The
-descriptors are not
-executable plans, and callers must not execute the first choice as a selection
-policy.
+descriptors are not executable plans, and callers must not execute the first
+choice as a selection policy.
 
 Opaque deployment extensions retain exact execution here; additional local
 alternatives require an explicit logical rule rather than a cost model making a