feat: preserve filters over planned summaries

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

@@ -1871,7 +1871,7 @@ fn keep_pre_asap_rc(expr: Rc<QueryExpr>) -> Result<Rc<SummaryNode>, ImplementErr
 /// `HAVING` predicate (the filter would need the estimate first), stays
 /// logical. Unsupported logical parents still conservatively become one
 /// [`SummaryExpr::KeepPreAsap`] subtree. Composable query-time value
-/// operators (`Project`, `Sort`, and `Limit`) are retained during final
+/// operators (`Project`, `Filter`, `Sort`, and `Limit`) are retained during final
 /// materialization so their independently planned children remain visible.
 pub fn bindable_intent(node: &QueryExpr) -> Option<&AggIntent> {
     if let QueryExpr::Aggregate {
@@ -3687,8 +3687,8 @@ impl<'a> GlobalSelection<'a> {
     /// Preserve composable query-time value operators in post-ASAP form even
     /// when the operator itself has no summary implementation. Its child is
     /// materialized independently, so a selected summary remains visible
-    /// beneath `Project`/`Sort`/`Limit` instead of being swallowed by one opaque
-    /// `KeepPreAsap` subtree.
+    /// beneath `Project`/`Filter`/`Sort`/`Limit` instead of being swallowed by
+    /// one opaque `KeepPreAsap` subtree.
     fn materialize_residual(
         &self,
         target: &Rc<QueryExpr>,
@@ -3705,6 +3705,9 @@ impl<'a> GlobalSelection<'a> {
                     qualifier: qualifier.clone(),
                 },
             ),
+            QueryExpr::Filter { pred, child } => {
+                (child, ValueOperation::Filter { pred: pred.clone() })
+            }
             QueryExpr::Sort {
                 keys,
                 partition_by,

crates/integration-tests/tests/sql_to_post_asap.rs

@@ -27,8 +27,9 @@ use asap_aware_mapping::{
 };
 use asap_frontend_sql::{lower_sql, SqlCatalog};
 use asap_types::post_asap::{
-    ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams,
-    SketchQuery, SummaryExpr, SummaryFamilyType, SummaryNode, SummarySchema, SummaryUpdate,
+    compile_executable_dag, ExactKind, ExactParams, ExecutableOperatorPayload, GroupingStrategy,
+    SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, SummaryFamilyType,
+    SummaryNode, SummarySchema, SummaryUpdate, ValueOperation,
 };
 use asap_types::pre_asap::expr_ir::ColumnRef;
 use asap_types::pre_asap::query_expr::{QueryExpr, Reduction};
@@ -152,6 +153,194 @@ async fn sql_full_query_retains_project_and_binds_inner_aggregate() {
     );
 }
 
+/// Relational parents emitted around a derived-table aggregate remain
+/// explicit read-time nodes while the aggregate is summary-bound.
+#[tokio::test]
+async fn sql_relational_parents_retain_summary_bound_aggregate() {
+    let pre_asap = Rc::new(
+        lower(
+            "SELECT t.service, t.p FROM \
+             (SELECT service, approx_percentile_cont(latency, 0.9) AS p \
+              FROM metrics GROUP BY service) t \
+             WHERE t.p > 100 ORDER BY t.p DESC LIMIT 5",
+            AccuracyTarget::Epsilon(0.01),
+        )
+        .await,
+    );
+    let space = search_workload(vec![("query", Rc::clone(&pre_asap))]);
+    let selection = space.global_selection(&DefaultCostModel);
+    let root = selection
+        .materialize(&space.roots[0].1)
+        .expect("materialization failed")
+        .expect("root must be discovered");
+
+    let mut node = root.as_ref();
+    let mut saw_project = false;
+    let mut saw_filter = false;
+    let mut saw_sort = false;
+    let mut saw_limit = false;
+    loop {
+        match &node.expr {
+            SummaryExpr::ValueOperation {
+                child, operation, ..
+            } => {
+                match operation {
+                    asap_types::post_asap::ValueOperation::Project { .. } => saw_project = true,
+                    asap_types::post_asap::ValueOperation::Filter { .. } => saw_filter = true,
+                    asap_types::post_asap::ValueOperation::Sort { .. } => saw_sort = true,
+                    asap_types::post_asap::ValueOperation::Limit { n, offset } => {
+                        assert_eq!((*n, *offset), (5, 0));
+                        saw_limit = true;
+                    }
+                    _ => {}
+                }
+                node = child;
+            }
+            SummaryExpr::SummaryEstimate { summary_input, .. } => {
+                assert!(matches!(summary_input.expr, SummaryExpr::SummaryAgg { .. }));
+                break;
+            }
+            other => panic!("expected relational parents over SummaryEstimate, got {other:?}"),
+        }
+    }
+    assert!(saw_project && saw_filter && saw_sort && saw_limit);
+}
+
+/// A post-aggregate predicate is read-time work, while a source predicate is
+/// part of the rows that populate the maintained summary. Neither predicate
+/// may be dropped or moved across the aggregation boundary.
+#[tokio::test]
+async fn sql_filter_keeps_read_predicate_and_summary_population_selection() {
+    let pre_asap = Rc::new(
+        lower(
+            "SELECT t.service, t.p FROM \
+             (SELECT service, approx_percentile_cont(latency, 0.9) AS p \
+              FROM metrics WHERE service = 'api' GROUP BY service) t \
+             WHERE t.p > 100",
+            AccuracyTarget::Epsilon(0.01),
+        )
+        .await,
+    );
+    let expected_read_predicate = {
+        let mut node = pre_asap.as_ref();
+        loop {
+            match node {
+                QueryExpr::Filter { pred, .. } => break pred.clone(),
+                QueryExpr::Project { child, .. }
+                | QueryExpr::Sort { child, .. }
+                | QueryExpr::Limit { child, .. } => node = child,
+                other => panic!("expected a Filter above the aggregate, got {other:?}"),
+            }
+        }
+    };
+    let expected_source_predicates = {
+        let mut node = pre_asap.as_ref();
+        loop {
+            match node {
+                QueryExpr::Scan { predicates, .. } => break predicates.clone(),
+                QueryExpr::Project { child, .. }
+                | QueryExpr::Filter { child, .. }
+                | QueryExpr::Aggregate { child, .. }
+                | QueryExpr::Sort { child, .. }
+                | QueryExpr::Limit { child, .. } => node = child,
+                other => panic!("expected a unary SQL plan over Scan, got {other:?}"),
+            }
+        }
+    };
+    assert_eq!(expected_source_predicates.len(), 1, "fixture source WHERE");
+
+    let space = search_workload(vec![("query", Rc::clone(&pre_asap))]);
+    let selection = space.global_selection(&DefaultCostModel);
+    let root = selection
+        .materialize(&space.roots[0].1)
+        .expect("materialization failed")
+        .expect("root must be discovered");
+
+    let mut node = root.as_ref();
+    let mut retained_read_predicate = None;
+    loop {
+        match &node.expr {
+            SummaryExpr::ValueOperation {
+                child,
+                operation: ValueOperation::Filter { pred },
+                ..
+            } => {
+                retained_read_predicate = Some(pred.clone());
+                node = child;
+            }
+            SummaryExpr::ValueOperation { child, .. } => node = child,
+            SummaryExpr::SummaryEstimate { summary_input, .. } => {
+                let SummaryExpr::SummaryAgg { child, .. } = &summary_input.expr else {
+                    panic!("expected SummaryAgg below SummaryEstimate");
+                };
+                let SummaryExpr::KeepPreAsap(raw_input) = &child.expr else {
+                    panic!("expected raw summary population below SummaryAgg");
+                };
+                let QueryExpr::Scan { predicates, .. } = raw_input.as_ref() else {
+                    panic!("expected source selection to remain a Scan");
+                };
+                assert_eq!(predicates, &expected_source_predicates);
+                break;
+            }
+            other => panic!("expected read-time operations over a summary, got {other:?}"),
+        }
+    }
+    assert_eq!(retained_read_predicate, Some(expected_read_predicate));
+
+    let executable = compile_executable_dag(&root).expect("typed DAG compilation failed");
+    assert!(executable.nodes.iter().any(|node| matches!(
+        &node.payload,
+        ExecutableOperatorPayload::Value {
+            operation: ValueOperation::Filter { pred },
+            ..
+        } if pred == retained_read_predicate.as_ref().unwrap()
+    )));
+}
+
+/// If the child has no legal summary implementation, retain only that child
+/// as the fallback leaf and keep the supported Filter as an explicit local
+/// read-time operation.
+#[tokio::test]
+async fn sql_filter_preserves_local_fallback_boundary_for_unsupported_child() {
+    let pre_asap = Rc::new(
+        lower(
+            "SELECT t.service, t.avg_bytes FROM \
+             (SELECT service, AVG(bytes) AS avg_bytes FROM metrics GROUP BY service) t \
+             WHERE t.avg_bytes > 100",
+            AccuracyTarget::Exact,
+        )
+        .await,
+    );
+    let space = search_workload(vec![("query", Rc::clone(&pre_asap))]);
+    let selection = space.global_selection(&DefaultCostModel);
+    let root = selection
+        .materialize(&space.roots[0].1)
+        .expect("materialization failed")
+        .expect("root must be discovered");
+
+    let mut node = root.as_ref();
+    let mut saw_filter = false;
+    loop {
+        match &node.expr {
+            SummaryExpr::ValueOperation {
+                child, operation, ..
+            } => {
+                saw_filter |= matches!(operation, ValueOperation::Filter { .. });
+                node = child;
+            }
+            SummaryExpr::KeepPreAsap(fallback) => {
+                assert!(
+                    matches!(fallback.as_ref(), QueryExpr::BinaryOp { .. }),
+                    "AVG's unsupported rewritten child should be opaque, got {fallback:?}"
+                );
+                break;
+            }
+            other => panic!("expected local value operations over fallback child, got {other:?}"),
+        }
+    }
+    assert!(saw_filter, "supported Filter must remain explicit");
+}
+
 /// `SELECT approx_percentile_cont(latency, 0.99) FROM metrics` at ε = 0.01,
 /// with the wrapping `Project` stripped (see module docs):
 ///

crates/types/src/post_asap/execution_data_state.rs

@@ -520,6 +520,7 @@ fn check_plain_operands(
         ValueOperation::Sort { .. }
             | ValueOperation::Limit { .. }
             | ValueOperation::Project { .. }
+            | ValueOperation::Filter { .. }
             | ValueOperation::FinalizeExactAccumulator
     ) {
         return check_plain_or_exact_values(input);

crates/types/src/post_asap/expr.rs

@@ -38,6 +38,12 @@ pub enum ValueOperation {
         cols: Vec<ProjectItem>,
         qualifier: Option<String>,
     },
+    /// Query-time row filtering. The predicate remains positional against
+    /// the child's output schema and is evaluated only after any summary
+    /// state below it has been read out to rows.
+    Filter {
+        pred: Predicate,
+    },
     /// Query-time ordering of the child's value rows. This is deliberately
     /// distinct from frequency-sketch heavy-hitter readout: PromQL `topk`
     /// ranks the values produced by its child at the evaluation timestamp.
@@ -93,18 +99,19 @@ pub struct SummaryNode {
 
 // ── Post-ASAP sketch-bound IR ────────────────────────────────────────────────
 
-/// Sketch-bound IR produced by post-ASAP binding rules. Those rules
-/// selectively replace logical aggregates and joins in the pre-ASAP
-/// `QueryExpr` with their summary-bound counterparts; everything not
-/// rewritten passes through as `KeepPreAsap(Rc<QueryExpr>)`.
+/// Sketch-bound IR produced by post-ASAP binding and final selection. Binding
+/// rules selectively replace logical aggregates and joins in the pre-ASAP
+/// `QueryExpr` with summary-bound counterparts. Final selection can retain
+/// supported read-time value operations around independently planned children;
+/// other unsupported subtrees pass through as `KeepPreAsap(Rc<QueryExpr>)`.
 ///
 /// Traversing from the root node yields a DAG; shared sub-expressions appear
 /// as multiple `Rc` references to the same `SummaryNode`.
 #[derive(Debug, Clone, PartialEq)]
 pub enum SummaryExpr {
-    /// A pre-ASAP subtree kept as-is — no binding rule rewrote it into
-    /// post-ASAP form (e.g. `Filter`, `Project`, `Sort`). Output schema is
-    /// the inner node's schema, lifted to `SummarySchema` with all fields as
+    /// A pre-ASAP subtree kept as-is because it has no selected implementation
+    /// or supported residual decomposition. Output schema is the inner node's
+    /// schema, lifted to `SummarySchema` with all fields as
     /// `SummaryFamilyType::Plain`.
     KeepPreAsap(Rc<QueryExpr>),