From f04cf822f6aac7ad2b17b468d855185f635ea7e5 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 4 Oct 2026 17:21:17 +0000 Subject: [PATCH 1/2] feat(frontend-sql): recognize the SQL frequency L2 and entropy idioms Port of the parked #562 and #564 for #509 Example 2. The SQL frontend names `SQRT(SUM(c*c))` over Float64 grouped unit counts as FrequencyL2, and `-SUM(p*LN(p))` with `p = COUNT(*)*1.0 / SUM(COUNT(*)) OVER ()` as FrequencyEntropy converted to nats. An exact population count guards SQL's empty-input NULL. Integer products, nullable keys, filters, HAVING, other log bases and partial windows are refused. The rules are the old Pass 1 SemanticEquivalentRewriteStrategy rules, run by lower_sql at the query root so the Stage 1 pipeline sees the intents; the exact candidate is the native exact reducer. The executor gains SQL sqrt (from #562). Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/executor/src/expressions/planner.rs | 24 +- crates/executor/tests/physical_semantics.rs | 30 ++ crates/frontend-sql/src/frequency.rs | 394 ++++++++++++++++++ crates/frontend-sql/src/lib.rs | 5 +- .../frontend-sql/tests/frequency_entropy.rs | 91 ++++ crates/frontend-sql/tests/frequency_l2.rs | 141 +++++++ crates/integration-tests/Cargo.toml | 1 + .../tests/sql_frequency_entropy.rs | 50 +++ .../tests/sql_frequency_l2.rs | 63 +++ 10 files changed, 798 insertions(+), 2 deletions(-) create mode 100644 crates/frontend-sql/src/frequency.rs create mode 100644 crates/frontend-sql/tests/frequency_entropy.rs create mode 100644 crates/frontend-sql/tests/frequency_l2.rs create mode 100644 crates/integration-tests/tests/sql_frequency_entropy.rs create mode 100644 crates/integration-tests/tests/sql_frequency_l2.rs diff --git a/Cargo.lock b/Cargo.lock index fcdc601d..6b00e464 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -397,6 +397,7 @@ name = "asap-integration-tests" version = "0.1.0" dependencies = [ "asap-executor", + "asap-frontend-common", "asap-frontend-promql", "asap-frontend-sql", "asap-logical-optimizer", diff --git a/crates/executor/src/expressions/planner.rs b/crates/executor/src/expressions/planner.rs index f877ff89..e4d73e05 100644 --- a/crates/executor/src/expressions/planner.rs +++ b/crates/executor/src/expressions/planner.rs @@ -127,6 +127,16 @@ pub(super) fn evaluate( .collect::, _>>()?; return Ok(Value::Float64(promql_function(name, &values)?)); } + if name.eq_ignore_ascii_case("sqrt") { + return match evaluate(&args[0], row, schema)? { + Value::Null => Ok(Value::Null), + Value::Float64(value) => Ok(Value::Float64(value.sqrt())), + Value::Int64(value) => Ok(Value::Float64((value as f64).sqrt())), + _ => Err(Error::Invalid( + "SQL sqrt requires a numeric argument".into(), + )), + }; + } if name == "promql_drop_metric_name" { let Value::Utf8(encoded) = evaluate(&args[0], row, schema)? else { return Err(Error::Invalid("series identity must be Utf8".into())); @@ -646,7 +656,19 @@ fn validate(expr: &ScalarExpr, schema: &planner_types::ir::schema::Schema) -> Re Ok(()) } ScalarExpr::FunctionCall { name, args } => { - if name != "promql_drop_metric_name" + if name.eq_ignore_ascii_case("sqrt") { + if args.len() != 1 + || !matches!( + args[0] + .scalar_type(schema) + .map_err(|e| Error::Invalid(e.to_string()))? + .0, + DataType::Int64 | DataType::Float64 | DataType::Null + ) + { + return Err(invalid()); + } + } else if name != "promql_drop_metric_name" && planner_types::ir::scalar::scalar_type_rules::promql_function_arity(name) .is_none() && name != "asap_struct_field" diff --git a/crates/executor/tests/physical_semantics.rs b/crates/executor/tests/physical_semantics.rs index 03a9b57a..e8e21e21 100644 --- a/crates/executor/tests/physical_semantics.rs +++ b/crates/executor/tests/physical_semantics.rs @@ -910,3 +910,33 @@ fn exact_cardinality_grouping_normalizes_float_identities() { [Value::Int64(2), Value::Int64(0)] )); } + +// SQL SQRT propagates NULL and accepts numeric inputs with a floating result. +#[test] +fn sql_sqrt_executes_numeric_and_null_arguments() { + for (dtype, value, expected) in [ + (DataType::Int64, Value::Int64(9), 3.0), + (DataType::Float64, Value::Float64(2.25), 1.5), + ] { + let input = schema(&[("v", dtype, true)]); + let expression = ScalarExpr::FunctionCall { + name: "sqrt".into(), + args: vec![ScalarExpr::Column(0)], + }; + let compiled = CompiledExpression::compile(&expression, &input).unwrap(); + assert!(matches!(compiled.evaluate(&[value]).unwrap(), Value::Float64(v) if v == expected)); + assert!(matches!( + compiled.evaluate(&[Value::Null]).unwrap(), + Value::Null + )); + } + let input = schema(&[("v", DataType::Float64, false)]); + let expression = ScalarExpr::FunctionCall { + name: "sqrt".into(), + args: vec![ScalarExpr::Column(0)], + }; + let compiled = CompiledExpression::compile(&expression, &input).unwrap(); + assert!( + matches!(compiled.evaluate(&[Value::Float64(-1.0)]).unwrap(), Value::Float64(v) if v.is_nan()) + ); +} diff --git a/crates/frontend-sql/src/frequency.rs b/crates/frontend-sql/src/frequency.rs new file mode 100644 index 00000000..c1b69b88 --- /dev/null +++ b/crates/frontend-sql/src/frequency.rs @@ -0,0 +1,394 @@ +//! Narrow frequency recognition over the resolved scalar/operator graph. +//! +//! #509 Example 2 writes `L2(x)` and `Entropy(x)` as SQL idioms over grouped +//! unit counts. These rules name them as `FrequencyL2` / `FrequencyEntropy` +//! intents, so Pass 1 can offer exact, summary and UnivMon realizations. +//! Projection lineage, predicates, accuracy, SQL's empty-input NULL and +//! entropy units (nats) are preserved; integer products and nullable keys +//! are refused because overflow and NULL groups are observable in SQL. +use asap_types::ir::operator::non_asap::any_measure_filtered; +use asap_types::ir::operator::{AggIntent, GroupKeys, Reduction}; +use asap_types::ir::scalar::{ArithmeticOpKind, CompareOpKind, ScalarValue}; +use asap_types::ir::schema::DataType; +use asap_types::ir::{ExprSemantics, NonASAPOp, Operator, OperatorNode, ProjectItem, ScalarExpr}; +use std::rc::Rc; + +/// The recognized form of a query whose root is the L2 or natural-log entropy +/// idiom over grouped unit counts, or `None` when neither rule applies. +pub fn recognize_frequency_idioms(root: &Rc) -> Option> { + frequency_entropy_rewrite(root).or_else(|| frequency_l2_rewrite(root)) +} + +// Only follow projections. Crossing a filter/limit would change the population. +fn expand( + mut expr: ScalarExpr, + mut node: Rc, +) -> Option<(ScalarExpr, Rc)> { + while let Some(NonASAPOp::Project { cols, child, .. }) = node.non_asap() { + expr = substitute(&expr, cols)?; + node = Rc::clone(child); + } + Some((expr, node)) +} +fn substitute(expr: &ScalarExpr, cols: &[ProjectItem]) -> Option { + Some(match expr { + ScalarExpr::Column(id) => cols.get(*id)?.expr.clone(), + ScalarExpr::Literal(_) => expr.clone(), + ScalarExpr::Negative { expr, semantics } => ScalarExpr::Negative { + expr: Box::new(substitute(expr, cols)?), + semantics: *semantics, + }, + ScalarExpr::Cast { + expr, + to, + try_cast: false, + } if *to == DataType::Float64 => ScalarExpr::Cast { + expr: Box::new(substitute(expr, cols)?), + to: to.clone(), + try_cast: false, + }, + ScalarExpr::Arithmetic { + op, + left, + right, + semantics, + } => ScalarExpr::Arithmetic { + op: op.clone(), + left: Box::new(substitute(left, cols)?), + right: Box::new(substitute(right, cols)?), + semantics: *semantics, + }, + ScalarExpr::FunctionCall { name, args } => ScalarExpr::FunctionCall { + name: name.clone(), + args: args + .iter() + .map(|arg| substitute(arg, cols)) + .collect::>()?, + }, + _ => return None, + }) +} +fn uncast(expr: &ScalarExpr) -> &ScalarExpr { + match expr { + ScalarExpr::Cast { + expr, + to: DataType::Float64, + try_cast: false, + } => uncast(expr), + _ => expr, + } +} + +fn frequency_l2_rewrite(root: &Rc) -> Option> { + let NonASAPOp::Project { cols, child, .. } = root.non_asap()? else { + return None; + }; + let [item] = cols.as_slice() else { + return None; + }; + let (expr, outer) = expand(item.expr.clone(), Rc::clone(child))?; + let ScalarExpr::FunctionCall { name, args } = &expr else { + return None; + }; + if !name.eq_ignore_ascii_case("sqrt") { + return None; + } + let [arg] = args.as_slice() else { + return None; + }; + if !matches!(uncast(arg), ScalarExpr::Column(0)) { + return None; + } + let NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures, + filters, + having: None, + child, + .. + } = outer.non_asap()? + else { + return None; + }; + if keys.is_without() || !keys.keys().is_empty() || any_measure_filtered(filters) { + return None; + } + let [AggIntent::Sum { col: Some(col) }] = measures.as_slice() else { + return None; + }; + let (product, inner) = expand(ScalarExpr::Column(*col), Rc::clone(child))?; + let ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Mul, + left, + right, + semantics: ExprSemantics::Sql, + } = &product + else { + return None; + }; + // SQL Int64 multiplication can overflow. Admit only products already typed Float64. + if product.scalar_type(&inner.schema).ok()?.0 != DataType::Float64 { + return None; + } + if !matches!( + (uncast(left), uncast(right)), + (ScalarExpr::Column(1), ScalarExpr::Column(1)) + ) { + return None; + } + let (key, accuracy, input) = grouped_unit_count(&inner)?; + sql_frequency_result( + root, + input, + AggIntent::FrequencyL2 { + col: Some(key), + accuracy, + }, + "frequency_l2", + ScalarExpr::Column(0), + ) +} + +// Both frequency rules require a complete, unfiltered count per non-NULL identity. +fn grouped_unit_count( + node: &Rc, +) -> Option<(usize, asap_types::types::AccuracyTarget, Rc)> { + let NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures, + filters, + having: None, + child, + .. + } = node.non_asap()? + else { + return None; + }; + let [key] = keys.keys() else { + return None; + }; + let [AggIntent::Count { accuracy }] = measures.as_slice() else { + return None; + }; + let field = child.schema.fields.get(*key)?; + if keys.is_without() + || any_measure_filtered(filters) + || field.nullable + || !matches!( + field.plain_dtype()?, + DataType::Bool | DataType::Int64 | DataType::Utf8 + ) + { + return None; + } + Some((*key, accuracy.clone(), Rc::clone(child))) +} + +fn probability_term(product: &ScalarExpr) -> Option<&ScalarExpr> { + let ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Mul, + left, + right, + semantics: ExprSemantics::Sql, + } = product + else { + return None; + }; + for (probability, logarithm) in [(left, right), (right, left)] { + let ScalarExpr::FunctionCall { name, args } = logarithm.as_ref() else { + continue; + }; + if name.eq_ignore_ascii_case("ln") && args.as_slice() == [probability.as_ref().clone()] { + return Some(probability); + } + } + None +} + +fn unit_count_term(expr: &ScalarExpr) -> bool { + match uncast(expr) { + ScalarExpr::Column(1) => true, + ScalarExpr::Arithmetic { op: ArithmeticOpKind::Mul, left, right, semantics: ExprSemantics::Sql } => { + [(left, right), (right, left)].into_iter().any(|(count, scale)| matches!(uncast(count), ScalarExpr::Column(1)) && matches!(scale.as_ref(), ScalarExpr::Literal(ScalarValue::Float64(value)) if *value == 1.0)) + }, + _ => false, + } +} + +fn frequency_entropy_rewrite(root: &Rc) -> Option> { + use asap_types::ir::operator::{WindowFrameBound, WindowFrameOffset, WindowFuncKind}; + let NonASAPOp::Project { cols, child, .. } = root.non_asap()? else { + return None; + }; + let [item] = cols.as_slice() else { + return None; + }; + let (expr, outer) = expand(item.expr.clone(), Rc::clone(child))?; + let ScalarExpr::Negative { + expr, + semantics: ExprSemantics::Sql, + } = expr + else { + return None; + }; + if !matches!(uncast(&expr), ScalarExpr::Column(0)) { + return None; + } + let NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures, + filters, + having: None, + child, + .. + } = outer.non_asap()? + else { + return None; + }; + if keys.is_without() || !keys.keys().is_empty() || any_measure_filtered(filters) { + return None; + } + let [AggIntent::Sum { col: Some(col) }] = measures.as_slice() else { + return None; + }; + let (product, window) = expand(ScalarExpr::Column(*col), Rc::clone(child))?; + let probability = probability_term(&product)?; + let ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Div, + left, + right, + semantics: ExprSemantics::Sql, + } = probability + else { + return None; + }; + if !unit_count_term(left) + || !matches!(uncast(right), ScalarExpr::Column(2)) + || probability.scalar_type(&window.schema).ok()?.0 != DataType::Float64 + { + return None; + } + let NonASAPOp::SQLWindowFunc { + func: WindowFuncKind::Sum, + args, + partition_by, + order_by, + frame: Some(frame), + child, + .. + } = window.non_asap()? + else { + return None; + }; + if partition_by.is_without() + || !partition_by.keys().is_empty() + || !order_by.is_empty() + || args.as_slice() != [ScalarExpr::Column(1)] + { + return None; + } + if !matches!( + frame.start_bound, + WindowFrameBound::Preceding(WindowFrameOffset::Scalar(ScalarValue::Null)) + ) || !matches!( + frame.end_bound, + WindowFrameBound::Following(WindowFrameOffset::Scalar(ScalarValue::Null)) + ) { + return None; + } + let (key, accuracy, input) = grouped_unit_count(child)?; + let nats = ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Mul, + left: Box::new(ScalarExpr::Column(0)), + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64( + std::f64::consts::LN_2, + ))), + semantics: ExprSemantics::Sql, + }; + // -SUM(p*LN(p)) is negative zero for a single-identity population. + let nats = ScalarExpr::Negative { + expr: Box::new(ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Sub, + left: Box::new(ScalarExpr::Literal(ScalarValue::Float64(0.0))), + right: Box::new(nats), + semantics: ExprSemantics::Sql, + }), + semantics: ExprSemantics::Sql, + }; + sql_frequency_result( + root, + input, + AggIntent::FrequencyEntropy { + col: Some(key), + accuracy, + }, + "frequency_entropy", + nats, + ) +} + +// Both rules use an exact population guard: an approximate statistic may be +// zero even for nonempty input, and must not control SQL's NULL result. +fn sql_frequency_result( + root: &Rc, + input: Rc, + measure: AggIntent, + name: &str, + value: ScalarExpr, +) -> Option> { + use asap_types::{ir::operator::JoinKind, ir::Predicate, types::AccuracyTarget}; + let NonASAPOp::Project { qualifier, .. } = root.non_asap()? else { + return None; + }; + let aggregate = |measure, name: &str| { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(GroupKeys::none()), + measures: vec![measure], + output_names: vec![name.into()], + filters: vec![], + having: None, + child: Rc::clone(&input), + })) + .ok() + }; + let statistic = aggregate(measure, name)?; + let count = aggregate( + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + "population_count", + )?; + let child = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Join { + kind: JoinKind::Cross, + pred: Predicate(ScalarExpr::Literal(ScalarValue::Boolean(true))), + left: statistic, + right: count, + })) + .ok()?; + let rewritten = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + cols: vec![ProjectItem { + alias: Some(root.schema.fields.first()?.name.clone()), + expr: ScalarExpr::Case { + operand: None, + branches: vec![( + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(1)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Int64(0))), + semantics: ExprSemantics::Sql, + }, + ScalarExpr::Cast { + expr: Box::new(ScalarExpr::Literal(ScalarValue::Null)), + to: DataType::Float64, + try_cast: false, + }, + )], + else_expr: Some(Box::new(value)), + }, + }], + qualifier: qualifier.clone(), + child, + })) + .ok()?; + (root.schema == rewritten.schema).then_some(rewritten) +} diff --git a/crates/frontend-sql/src/lib.rs b/crates/frontend-sql/src/lib.rs index 4d9e6ea1..64e3df7c 100644 --- a/crates/frontend-sql/src/lib.rs +++ b/crates/frontend-sql/src/lib.rs @@ -9,6 +9,7 @@ //! Depends on DataFusion only — never on the PromQL parser. pub mod error; +pub mod frequency; pub mod sql; use std::rc::Rc; @@ -55,7 +56,9 @@ pub async fn lower_sql_dialect( // Binding resolves names and derives every node's schema; result-type // checks (such as temporal subtraction, whose duration unit the IR cannot // represent) surface here as `ResolveDAGError::Schema`. - Ok(resolve_root(&unresolved)?) + let root = resolve_root(&unresolved)?; + // #509 Example 2: the L2 and entropy idioms become frequency intents. + Ok(frequency::recognize_frequency_idioms(&root).unwrap_or(root)) } /// Lower every SQL batch entry in `workload` to an operator DAG. diff --git a/crates/frontend-sql/tests/frequency_entropy.rs b/crates/frontend-sql/tests/frequency_entropy.rs new file mode 100644 index 00000000..ea92e23f --- /dev/null +++ b/crates/frontend-sql/tests/frequency_entropy.rs @@ -0,0 +1,91 @@ +//! The SQL frontend names the natural-log entropy idiom as a `FrequencyEntropy` intent. +use asap_frontend_common::resolve_root; +use asap_frontend_sql::{lower_sql, SqlCatalog, SqlLowerer}; +use asap_types::{ + ir::operator::AggIntent, + ir::schema::{DataType, Field, Schema}, + ir::{NonASAPOp, OperatorNode}, + types::AccuracyTarget, +}; +fn catalog(nullable: bool) -> SqlCatalog { + SqlCatalog::new().with_table( + "flows", + Schema::new(vec![Field::plain("src_ip", DataType::Utf8, nullable)]), + ) +} +fn has_entropy(node: &OperatorNode) -> bool { + if let Some(NonASAPOp::Aggregate { measures, .. }) = node.non_asap() { + if measures + .iter() + .any(|m| matches!(m, AggIntent::FrequencyEntropy { .. })) + { + return true; + } + } + node.children().iter().any(|child| has_entropy(child)) +} +// The proposal's natural-log idiom lowers to an entropy intent and preserves output schema. +#[tokio::test] +async fn recognizes_sql_entropy_in_nats() { + let sql = "SELECT -SUM(p * LN(p)) AS entropy FROM (SELECT COUNT(*) * 1.0 / SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip) f"; + let rewritten = lower_sql(sql, &catalog(false), AccuracyTarget::Exact) + .await + .unwrap(); + let original = resolve_root( + &SqlLowerer::new(&catalog(false)) + .lower(sql, &AccuracyTarget::Exact) + .await + .unwrap(), + ) + .unwrap(); + assert!(has_entropy(&rewritten)); + assert_eq!(original.schema, rewritten.schema); + assert!(!has_entropy(&original)); +} + +// Changes to units, normalization, window coverage or the counted population are not entropy rewrites. +#[tokio::test] +async fn declines_non_equivalent_entropy_shapes() { + for (sql, nullable) in [ + ("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*2.0/SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip) f", false), + ("SELECT -SUM(p*LOG2(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip) f", false), + ("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER (PARTITION BY src_ip) AS p FROM flows GROUP BY src_ip) f", false), + ("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER (ORDER BY src_ip) AS p FROM flows GROUP BY src_ip) f", false), + ("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip HAVING COUNT(*) > 1) f", false), + ("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip) f", true), + ] { + let root = lower_sql(sql, &catalog(nullable), AccuracyTarget::Exact).await.unwrap(); + assert!(!has_entropy(&root)); + } +} + +// Entropy receives the source query budget while its empty-population guard stays exact. +#[tokio::test] +async fn entropy_accuracy_and_population_guard_are_separate() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + }; + let node = lower_sql("SELECT -SUM(p*LN(p)) FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER () AS p FROM flows GROUP BY src_ip) f", &catalog(false), target.clone()).await.unwrap(); + let NonASAPOp::Project { child, .. } = node.expect_non_asap() else { + panic!("project"); + }; + let NonASAPOp::Join { left, right, .. } = child.expect_non_asap() else { + panic!("guarded statistic"); + }; + let NonASAPOp::Aggregate { measures, .. } = left.expect_non_asap() else { + panic!("entropy"); + }; + assert!( + matches!(&measures[0], AggIntent::FrequencyEntropy { accuracy, .. } if *accuracy == target) + ); + let NonASAPOp::Aggregate { measures, .. } = right.expect_non_asap() else { + panic!("count"); + }; + assert!(matches!( + measures.as_slice(), + [AggIntent::Count { + accuracy: AccuracyTarget::Exact + }] + )); +} diff --git a/crates/frontend-sql/tests/frequency_l2.rs b/crates/frontend-sql/tests/frequency_l2.rs new file mode 100644 index 00000000..43bd934d --- /dev/null +++ b/crates/frontend-sql/tests/frequency_l2.rs @@ -0,0 +1,141 @@ +//! The SQL frontend names the frequency L2 idiom as a `FrequencyL2` intent. +use asap_frontend_common::resolve_root; +use asap_frontend_sql::{lower_sql, SqlCatalog, SqlLowerer}; +use asap_types::{ + ir::operator::AggIntent, + ir::schema::{DataType, Field, Schema}, + ir::{NonASAPOp, OperatorNode}, + types::AccuracyTarget, +}; +use std::rc::Rc; + +fn catalog(nullable: bool) -> SqlCatalog { + SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("src_ip", DataType::Utf8, nullable), + Field::plain("keep", DataType::Bool, false), + ]), + ) +} +fn has_l2(node: &OperatorNode) -> bool { + if let Some(NonASAPOp::Aggregate { measures, .. }) = node.non_asap() { + if measures + .iter() + .any(|m| matches!(m, AggIntent::FrequencyL2 { .. })) + { + return true; + } + } + node.children().iter().any(|child| has_l2(child)) +} +/// The relational DAG the frontend lowers before frequency recognition. +async fn relational(sql: &str, catalog: &SqlCatalog) -> Rc { + resolve_root( + &SqlLowerer::new(catalog) + .lower(sql, &AccuracyTarget::Exact) + .await + .unwrap(), + ) + .unwrap() +} +// Floating count products retain aliases, filters, type and empty-input nullability. +#[tokio::test] +async fn recognizes_float_frequency_l2() { + for sql in [ + "SELECT SQRT(SUM(CAST(c AS DOUBLE) * CAST(c AS DOUBLE))) AS norm FROM (SELECT src_ip, COUNT(*) AS c FROM flows WHERE keep GROUP BY src_ip) f", + "SELECT SQRT(SUM(c * c)) AS norm FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", + ] { + let rewritten = lower_sql(sql, &catalog(false), AccuracyTarget::Exact).await.unwrap(); + let original = relational(sql, &catalog(false)).await; + assert!(has_l2(&rewritten)); + assert_eq!(original.schema, rewritten.schema); + assert!(!has_l2(&original)); + } +} +// Nearby SQL forms with different frequency, overflow or NULL semantics remain ordinary SQL. +#[tokio::test] +async fn declines_non_equivalent_frequency_shapes() { + for (sql, nullable) in [ + ("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, COUNT(*) AS c FROM flows GROUP BY src_ip) f", false), + ("SELECT SQRT(SUM(CAST(c AS DOUBLE)*CAST(c AS DOUBLE))) FROM (SELECT src_ip, COUNT(*) AS c FROM flows GROUP BY src_ip) f", true), + ("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip HAVING COUNT(*) > 1) f", false), + ("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(SUM(CAST(keep AS BIGINT)) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", false), + ] { + let root = lower_sql(sql, &catalog(nullable), AccuracyTarget::Exact).await.unwrap(); + assert!(!has_l2(&root)); + } +} + +// The new frequency intent carries the query's requested budget rather than an invented default. +#[tokio::test] +async fn frequency_l2_preserves_accuracy_target() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let node = lower_sql("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", &catalog(false), target.clone()).await.unwrap(); + let NonASAPOp::Project { child, .. } = node.expect_non_asap() else { + panic!("project"); + }; + let NonASAPOp::Join { left, .. } = child.expect_non_asap() else { + panic!("guarded statistic"); + }; + let NonASAPOp::Aggregate { measures, .. } = left.expect_non_asap() else { + panic!("aggregate"); + }; + assert!(matches!(&measures[0], AggIntent::FrequencyL2 { accuracy, .. } if *accuracy == target)); +} + +// Stage 1 offers the exact L2 reducer and summary alternatives for the recognized intent. +#[tokio::test] +async fn stage1_offers_exact_and_summary_l2_alternatives() { + use asap_logical_optimizer::pass1::logical_candidates::enumerate_local_logical_candidates; + use asap_logical_optimizer::pass1::replacement::Realization; + use asap_types::ir::QueryRoot; + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let root = lower_sql("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", &catalog(false), target).await.unwrap(); + let inventory = + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap(); + let l2 = inventory + .targets + .iter() + .find(|t| has_l2(&t.target)) + .expect("L2 target"); + assert!(l2 + .alternatives + .iter() + .any(|a| matches!(a, Realization::PassThrough))); + assert!(l2 + .alternatives + .iter() + .any(|a| matches!(a, Realization::Sketch(_)))); +} + +// An approximate L2 estimate must never decide whether SQL returns NULL. +#[tokio::test] +async fn frequency_empty_input_guard_uses_an_exact_count() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let node = lower_sql("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", &catalog(false), target).await.unwrap(); + let NonASAPOp::Project { child, .. } = node.expect_non_asap() else { + panic!("project"); + }; + let NonASAPOp::Join { right, .. } = child.expect_non_asap() else { + panic!("exact population guard"); + }; + let NonASAPOp::Aggregate { measures, .. } = right.expect_non_asap() else { + panic!("count"); + }; + assert!(matches!( + measures.as_slice(), + [AggIntent::Count { + accuracy: AccuracyTarget::Exact + }] + )); +} diff --git a/crates/integration-tests/Cargo.toml b/crates/integration-tests/Cargo.toml index 0d2f938b..6f43455c 100644 --- a/crates/integration-tests/Cargo.toml +++ b/crates/integration-tests/Cargo.toml @@ -7,6 +7,7 @@ edition = "2021" asap-types = { path = "../types" } asap-frontend-promql = { path = "../frontend-promql" } asap-frontend-sql = { path = "../frontend-sql" } +asap-frontend-common = { path = "../frontend-common" } asap-plan-selection = { path = "../plan-selection" } asap-logical-optimizer = { path = "../logical-optimizer" } asap-physical-optimizer = { path = "../physical-optimizer" } diff --git a/crates/integration-tests/tests/sql_frequency_entropy.rs b/crates/integration-tests/tests/sql_frequency_entropy.rs new file mode 100644 index 00000000..a2b61c74 --- /dev/null +++ b/crates/integration-tests/tests/sql_frequency_entropy.rs @@ -0,0 +1,50 @@ +//! The proposal's entropy idiom executes in native exact frequency operators, with SQL units. +mod physical_common; +use asap_executor::values::Value; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_types::{ + ir::schema::{DataType, Field, Schema}, + types::AccuracyTarget, +}; + +// Native execution produces nats, NULL for no population and SQL's negative zero for one identity. +#[tokio::test] +async fn entropy_rewrite_executes_nats_and_empty_population_guard() { + let catalog = SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("src_ip", DataType::Utf8, false), + Field::plain("keep", DataType::Bool, false), + ]), + ); + let sql = "SELECT -SUM(p*LN(p)) AS entropy FROM (SELECT COUNT(*)*1.0/SUM(COUNT(*)) OVER () AS p FROM flows WHERE keep GROUP BY src_ip) f"; + let rewritten = lower_sql(sql, &catalog, AccuracyTarget::Exact) + .await + .unwrap(); + for (keys, expected) in [ + (vec![], None), + (vec!["a", "a"], Some(-0.0)), + (vec!["a", "a", "b", "b"], Some(std::f64::consts::LN_2)), + ( + vec!["a", "a", "a", "b"], + Some(-0.75_f64 * 0.75_f64.ln() - 0.25_f64 * 0.25_f64.ln()), + ), + ] { + let mut rows: Vec<_> = keys + .into_iter() + .map(|key| vec![Value::Utf8(key.into()), Value::Bool(true)]) + .collect(); + rows.push(vec![Value::Utf8("discard".into()), Value::Bool(false)]); + let actual = physical_common::execute_raw_rows(&rewritten, rows); + match (&actual[0][0], expected) { + (Value::Null, None) => {} + (Value::Float64(value), Some(expected)) => { + assert!((value - expected).abs() < 1e-12); + if expected == 0.0 { + assert_eq!(value.to_bits(), (-0.0_f64).to_bits()); + } + } + other => panic!("wrong SQL entropy: {other:?}"), + } + } +} diff --git a/crates/integration-tests/tests/sql_frequency_l2.rs b/crates/integration-tests/tests/sql_frequency_l2.rs new file mode 100644 index 00000000..ab3e68e2 --- /dev/null +++ b/crates/integration-tests/tests/sql_frequency_l2.rs @@ -0,0 +1,63 @@ +//! SQL L2 recognition survives physical compilation and executes the exact native reducer. +mod physical_common; +use asap_executor::values::Value; +use asap_frontend_common::resolve_root; +use asap_frontend_sql::{lower_sql, SqlCatalog, SqlLowerer}; +use asap_types::{ + ir::schema::{DataType, Field, Schema}, + ir::NonASAPOp, + types::AccuracyTarget, +}; + +// Original SQL and its recognized form agree on filters and SQL's empty-input NULL. +#[tokio::test] +async fn sql_l2_original_and_rewrite_execute_equivalently() { + let catalog = SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("src_ip", DataType::Utf8, false), + Field::plain("keep", DataType::Bool, false), + ]), + ); + let sql = "SELECT SQRT(SUM(CAST(c AS DOUBLE)*CAST(c AS DOUBLE))) AS norm FROM (SELECT src_ip, COUNT(*) AS c FROM flows WHERE keep GROUP BY src_ip) f"; + let root = resolve_root( + &SqlLowerer::new(&catalog) + .lower(sql, &AccuracyTarget::Exact) + .await + .unwrap(), + ) + .unwrap(); + let rewritten = lower_sql(sql, &catalog, AccuracyTarget::Exact) + .await + .unwrap(); + assert!(matches!( + rewritten.non_asap(), + Some(NonASAPOp::Project { .. }) + )); + for rows in [ + vec![], + vec![vec![Value::Utf8("discard".into()), Value::Bool(false)]], + vec![ + vec![Value::Utf8("a".into()), Value::Bool(true)], + vec![Value::Utf8("a".into()), Value::Bool(true)], + vec![Value::Utf8("b".into()), Value::Bool(true)], + vec![Value::Utf8("discard".into()), Value::Bool(false)], + ], + ] { + let original = physical_common::execute_raw_rows(&root, rows.clone()); + let actual = physical_common::execute_raw_rows(&rewritten, rows); + if original.iter().any(|row| !matches!(row[0], Value::Null)) { + assert!( + matches!(original[0][0], Value::Float64(v) if (v - 5.0_f64.sqrt()).abs() < 1e-12) + ); + } + assert_eq!(original.len(), actual.len()); + for (expected, actual) in original.iter().zip(actual) { + match (&expected[0], &actual[0]) { + (Value::Null, Value::Null) => {} + (Value::Float64(a), Value::Float64(b)) => assert!((a - b).abs() < 1e-12), + other => panic!("mismatched SQL result: {other:?}"), + } + } + } +} From fc32eae91a96774f829191d7ab6b4e3508647bfe Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 4 Oct 2026 18:23:26 +0000 Subject: [PATCH 2/2] fix: integrate with #593 Pass 1 takes the declared metric types since #593. Co-Authored-By: Claude Opus 5.5 --- crates/frontend-sql/tests/frequency_l2.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/frontend-sql/tests/frequency_l2.rs b/crates/frontend-sql/tests/frequency_l2.rs index 43bd934d..ff4c0563 100644 --- a/crates/frontend-sql/tests/frequency_l2.rs +++ b/crates/frontend-sql/tests/frequency_l2.rs @@ -98,8 +98,11 @@ async fn stage1_offers_exact_and_summary_l2_alternatives() { delta: 0.01, }; let root = lower_sql("SELECT SQRT(SUM(c*c)) FROM (SELECT src_ip, CAST(COUNT(*) AS DOUBLE) AS c FROM flows GROUP BY src_ip) f", &catalog(false), target).await.unwrap(); - let inventory = - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap(); + let inventory = enumerate_local_logical_candidates( + vec![(0, QueryRoot::Operator(root))], + &Default::default(), + ) + .unwrap(); let l2 = inventory .targets .iter()