From 277bc65f4bb8e38c26f008438eb53119c04a8e49 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sat, 3 Oct 2026 15:06:51 +0000 Subject: [PATCH] feat(planner): recognize floating SQL frequency L2 idioms --- .../src/frequency_rewrite.rs | 198 ++++++++++++++++++ crates/asap-aware-mapping/src/lib.rs | 1 + crates/asap-aware-mapping/src/rewrite.rs | 17 ++ .../src/expressions/planner.rs | 24 ++- .../tests/physical_semantics.rs | 30 +++ crates/frontend-sql/tests/frequency_l2.rs | 102 +++++++++ .../tests/sql_frequency_l2.rs | 123 +++++++++++ docs/develop_docs/planner-layering-status.md | 18 +- 8 files changed, 511 insertions(+), 2 deletions(-) create mode 100644 crates/asap-aware-mapping/src/frequency_rewrite.rs create mode 100644 crates/frontend-sql/tests/frequency_l2.rs create mode 100644 crates/integration-tests/tests/sql_frequency_l2.rs diff --git a/crates/asap-aware-mapping/src/frequency_rewrite.rs b/crates/asap-aware-mapping/src/frequency_rewrite.rs new file mode 100644 index 000000000..1ab027ea4 --- /dev/null +++ b/crates/asap-aware-mapping/src/frequency_rewrite.rs @@ -0,0 +1,198 @@ +//! Narrow frequency recognition over the resolved scalar/operator graph. +use asap_types::ir::non_asap::any_measure_filtered; +use asap_types::{ + ir::{ExprSemantics, NonASAPOp, Operator, OperatorNode, ProjectItem, ScalarExpr}, + pre_asap::{ + AggIntent, ArithmeticOpKind, CompareOpKind, DataType, GroupKeys, Reduction, ScalarValue, + }, +}; +use std::rc::Rc; + +// 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::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, + } +} + +pub(super) fn frequency_l2_rewrite(root: &Rc) -> Option> { + let NonASAPOp::Project { + cols, + child, + qualifier, + } = 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; + } + let NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures, + filters, + having: None, + child, + .. + } = inner.non_asap()? + else { + return None; + }; + let [key] = keys.keys() else { + return None; + }; + if keys.is_without() || any_measure_filtered(filters) { + return None; + } + let [AggIntent::Count { accuracy }] = measures.as_slice() else { + return None; + }; + if !matches!( + (uncast(left), uncast(right)), + (ScalarExpr::Column(1), ScalarExpr::Column(1)) + ) { + return None; + } + let field = child.schema.fields.get(*key)?; + // COUNT(*) GROUP BY NULL creates a real group; the frequency intent skips it. + if field.nullable + || !matches!( + field.plain_dtype()?, + DataType::Bool | DataType::Int64 | DataType::Utf8 + ) + { + return None; + } + let aggregate = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(GroupKeys::none()), + measures: vec![AggIntent::FrequencyL2 { + col: Some(*key), + accuracy: accuracy.clone(), + }], + output_names: vec!["frequency_l2".into()], + filters: vec![], + having: None, + child: Rc::clone(child), + })) + .ok()?; + // L2 is positive for any nonempty unit-update population. Restore SQL SUM's + // NULL on an empty relation without introducing another count computation. + 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(0)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(0.0))), + semantics: ExprSemantics::Sql, + }, + ScalarExpr::Cast { + expr: Box::new(ScalarExpr::Literal(ScalarValue::Null)), + to: DataType::Float64, + try_cast: false, + }, + )], + else_expr: Some(Box::new(ScalarExpr::Column(0))), + }, + }], + qualifier: qualifier.clone(), + child: aggregate, + })) + .ok()?; + (root.schema == rewritten.schema).then_some(rewritten) +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 16e6e6118..493042499 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -159,6 +159,7 @@ pub mod empirical_resources; pub mod erp; pub mod exact_composition; pub mod explanation; +mod frequency_rewrite; mod function_rules; pub mod grouping; pub mod pane_sharing; diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index ca752a971..21864c8b4 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -26,6 +26,14 @@ //! that reshaping — see "Non-goals" below for why it does not also decide //! whether the reshaping is worth it. //! +//! ## SQL frequency recognition +//! +//! A floating-point `SQRT(SUM(c*c))` over grouped unit counts exposes a +//! frequency L2 alternative through the same semantic strategy. Projection +//! lineage, predicates, accuracy and empty-input NULL are preserved. Integer +//! products and nullable grouping keys are excluded because overflow and NULL +//! groups have observable SQL behavior. See `frequency_rewrite` for the rule. +//! //! ## Scope //! //! Ordinary `by(...)` averages use a schema-preserving projection. Temporal @@ -399,9 +407,18 @@ impl ReplacementStrategy for SemanticEquivalentRewriteStrategy { fn matches(&self, target: &TargetSubDAG<'_>) -> bool { avg_rewrite_target(target.root).is_some() || composed_aggregate_rewrite(target.root).is_some() + || crate::frequency_rewrite::frequency_l2_rewrite(target.root).is_some() } fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + if let Some(rewritten) = crate::frequency_rewrite::frequency_l2_rewrite(target.root) { + return vec![ReplacementSubDAG { + strategy: "SemanticEquivalentRewriteStrategy", + replacement: Replacement::SubDAG(rewritten), + provenance: crate::replacement::ReplacementProvenance::LogicalRewrite, + rationale: "recognize a floating SQL frequency L2 product while preserving empty-input NULL and the original exact candidate".into(), + }]; + } if let Some(rewritten) = composed_aggregate_rewrite(target.root) { return vec![ReplacementSubDAG { strategy: "SemanticEquivalentRewriteStrategy", diff --git a/crates/asap-physical-operators/src/expressions/planner.rs b/crates/asap-physical-operators/src/expressions/planner.rs index 99f4c2b89..11bea51ae 100644 --- a/crates/asap-physical-operators/src/expressions/planner.rs +++ b/crates/asap-physical-operators/src/expressions/planner.rs @@ -126,6 +126,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())); @@ -645,7 +655,19 @@ fn validate(expr: &ScalarExpr, schema: &planner_types::pre_asap::Schema) -> Resu 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::pre_asap::scalar_type_rules::promql_function_arity(name).is_none() && name != "asap_struct_field" && name != "asap_element_access" diff --git a/crates/asap-physical-operators/tests/physical_semantics.rs b/crates/asap-physical-operators/tests/physical_semantics.rs index 5939be253..815108f1f 100644 --- a/crates/asap-physical-operators/tests/physical_semantics.rs +++ b/crates/asap-physical-operators/tests/physical_semantics.rs @@ -831,3 +831,33 @@ fn exact_frequency_grouping_and_entropy_bits() { } assert!(unary(input, vec![], operator).is_empty()); } + +// 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 = QueryExpr::FunctionCall { + name: "sqrt".into(), + args: vec![QueryExpr::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 = QueryExpr::FunctionCall { + name: "sqrt".into(), + args: vec![QueryExpr::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/tests/frequency_l2.rs b/crates/frontend-sql/tests/frequency_l2.rs new file mode 100644 index 000000000..8c32b6340 --- /dev/null +++ b/crates/frontend-sql/tests/frequency_l2.rs @@ -0,0 +1,102 @@ +//! SQL frequency idioms expose logical candidates without replacing the exact SQL DAG. +use asap_aware_mapping::{ + replacement::{Replacement, ReplacementStrategy, TargetSubDAG}, + SemanticEquivalentRewriteStrategy, +}; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_types::{ + ir::{NonASAPOp, OperatorNode}, + pre_asap::{AggIntent, DataType, Field, Schema}, + types::AccuracyTarget, +}; + +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)) +} +// Floating count products retain aliases, filters, type and empty-input nullability. +#[tokio::test] +async fn recognizes_float_frequency_l2_as_an_additional_candidate() { + 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 root = lower_sql(sql, &catalog(false), AccuracyTarget::Exact).await.unwrap(); + let replacements = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); + let rewritten = replacements.iter().find_map(|r| match &r.replacement { Replacement::SubDAG(node) if has_l2(node) => Some(node), _ => None }).expect("frequency L2 candidate"); + assert_eq!(root.schema, rewritten.schema); + assert!(!has_l2(&root)); + } +} +// 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(); + let candidates = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); + assert!(!candidates.iter().any(|r| matches!(&r.replacement, Replacement::SubDAG(n) if has_l2(n)))); + } +} + +// 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 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.clone()).await.unwrap(); + let candidates = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); + let Replacement::SubDAG(node) = &candidates[0].replacement else { + panic!("rewrite"); + }; + let NonASAPOp::Project { child, .. } = node.expect_non_asap() else { + panic!("project"); + }; + let NonASAPOp::Aggregate { measures, .. } = child.expect_non_asap() else { + panic!("aggregate"); + }; + assert!(matches!(&measures[0], AggIntent::FrequencyL2 { accuracy, .. } if *accuracy == target)); +} + +// The default search discovers the rewrite and keeps an original relational alternative. +#[tokio::test] +async fn default_search_keeps_exact_sql_and_frequency_alternatives() { + use asap_aware_mapping::replacement::{default_strategies, search_workload_with_targets}; + 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), AccuracyTarget::Exact).await.unwrap(); + let space = search_workload_with_targets( + vec![(0, root, Some(AccuracyTarget::Exact))], + &default_strategies(), + &asap_aware_mapping::accuracy::DefaultAccuracyModel, + ); + let inventory = space.enumerate_candidate_dags(1000).unwrap(); + assert!(inventory + .candidates + .iter() + .any(|candidate| has_l2(&candidate[0].1))); + assert!(inventory + .candidates + .iter() + .any(|candidate| !has_l2(&candidate[0].1))); +} 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 000000000..f7bea5760 --- /dev/null +++ b/crates/integration-tests/tests/sql_frequency_l2.rs @@ -0,0 +1,123 @@ +//! SQL L2 recognition survives wire compilation and executes the exact native fallback. +mod physical_common; +use asap_aware_mapping::{ + replacement::{Replacement, ReplacementStrategy, TargetSubDAG}, + SemanticEquivalentRewriteStrategy, +}; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_physical_operators::{ + runtime::Scope, + values::{Batch, Value}, +}; +use asap_types::{ + ir::{ + export::{NonASAPOpKind, PostAsapOperatorPayload}, + NonASAPOp, OperatorNode, + }, + pre_asap::{DataType, Field, Schema}, + types::AccuracyTarget, +}; +use std::{collections::BTreeMap, rc::Rc, sync::Arc}; + +fn run(root: &Rc, rows: Vec>) -> Vec> { + use asap_physical_operators::{ + physical_planner::bind_with_data_sources, + runtime::{Limits, RunContext}, + sources::{DataSources, MemorySource}, + }; + use futures::{executor::block_on, StreamExt}; + let wire = physical_common::compile_post_asap_dag(root).unwrap(); + let scan = wire + .nodes + .iter() + .find(|node| { + matches!( + node.payload, + PostAsapOperatorPayload::Relational { + operator: NonASAPOpKind::Scan { .. } + } + ) + }) + .unwrap(); + let PostAsapOperatorPayload::Relational { + operator: NonASAPOpKind::Scan { source, .. }, + } = &scan.payload + else { + unreachable!(); + }; + let input = Arc::new(scan.output_schema.clone()); + let batch = Batch::try_new(input.clone(), rows).unwrap(); + let mut sources = DataSources::default(); + sources + .register( + source.clone(), + Arc::new(MemorySource::new(input, vec![batch]).unwrap()), + ) + .unwrap(); + let root_id = u64::from(wire.root.0); + let plan = bind_with_data_sources(&wire, BTreeMap::new(), &[root_id], &sources).unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let result = block_on(async { + let mut output = plan.execute(&[root_id], context.clone()).unwrap().remove(0); + let mut rows = vec![]; + while let Some(batch) = output.next().await { + rows.extend_from_slice(batch.unwrap().rows()); + } + rows + }); + assert_eq!(context.retained_bytes(), 0); + result +} +// Original SQL and its logical alternative 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 root = lower_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", &catalog, AccuracyTarget::Exact).await.unwrap(); + let replacements = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); + let Replacement::SubDAG(rewritten) = &replacements[0].replacement else { + panic!("logical rewrite"); + }; + 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 = run(&root, rows.clone()); + let actual = run(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:?}"), + } + } + } +} diff --git a/docs/develop_docs/planner-layering-status.md b/docs/develop_docs/planner-layering-status.md index 131f2b124..f2df5ec29 100644 --- a/docs/develop_docs/planner-layering-status.md +++ b/docs/develop_docs/planner-layering-status.md @@ -6,7 +6,7 @@ is a target contract, not a statement that its examples execute today. | Proposal contract | Evidence at #557 | Remaining scope | | --- | --- | --- | -| Language frontends and common logical IR | SQL/PromQL/MetricsQL lower to unified operators and scalars. | Example 2 SQL frequency L2 and entropy idioms are not recognized. Preserve alias lineage, filters, NULL groups, empty inputs, count overflow and entropy units when adding recognition. | +| Language frontends and common logical IR | SQL/PromQL/MetricsQL lower to unified operators and scalars. | Floating-point SQL frequency L2 products now have a conservative logical rewrite. Integer products and the entropy idiom remain unrecognized. Preserve alias lineage, filters, NULL groups, empty inputs, count overflow and entropy units when adding recognition. | | Local exact and summary alternatives | `replacement::summary_candidates`, realization rules and candidate inventory exist; supplied accuracy models reach Pass 1. | Specialized entropy/norm families in Example 2 are illustrative, not registered families. UnivMon certifies only unit-update total count; L2, entropy and cardinality epsilon/delta bounds need verified evidence or a deployment model. | | Summary-capability sharing | CSE interns structurally identical producers, including states with different readers. | It does not enumerate all partial sharing partitions or resize compatible states to the strictest consumer. Example 2's 37 candidates are not an acceptance result. | | Window composition | Mergeable state IR/native merge exists; physical pane compatibility and reuse cost helpers exist. | Automatic logical sliding/tumbling/EH alternatives over differing windows, boundary coverage and error proofs are absent. A merge kernel alone does not implement Examples 1/3. | @@ -45,3 +45,19 @@ is a target contract, not a statement that its examples execute today. The examples' numerical candidate counts depend on their stated rule sets. Tests should establish those rule sets explicitly before asserting the counts. + +## SQL L2 follow-up acceptance + +`SemanticEquivalentRewriteStrategy` recognizes a single `SQRT(SUM(c*c))` +output when `c` is a grouped unit count and the product is already Float64. +It follows positional projection lineage, keeps the input predicates, inherits +count accuracy, and restores SQL's NULL result for an empty population. It +retains the original exact candidate. Recognition requires one nonnullable +Boolean, Int64 or Utf8 grouping key, no measure filters/HAVING and no +intervening operators that change the grouped population. The uncast integer +product in Example 2 remains a gap because SQL overflow is observable. + +`frontend-sql/tests/frequency_l2.rs` covers recognition, refusal boundaries, +accuracy propagation and candidate retention; `integration-tests/tests/sql_frequency_l2.rs` +executes both SQL and the rewrite through raw connectors and wire compilation. +This step does not supply an L2 accuracy certificate or all sharing partitions.