diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 84851870..03da4fa8 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1871,7 +1871,7 @@ fn keep_pre_asap_rc(expr: Rc) -> Result, 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, @@ -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, diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index fba42065..0d1f9843 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/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): /// diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 85ad1568..76478f40 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/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); diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index d9c91454..87944a24 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -38,6 +38,12 @@ pub enum ValueOperation { cols: Vec, qualifier: Option, }, + /// 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)`. +/// 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)`. /// /// 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),