feat(planner): recognize floating SQL frequency L2 idioms

[zzylol]

Problem: SQL SQRT(SUM(c*c)) over grouped counts is never seen as L2(x)

#509 §Example 2 writes Q3 as an SQL idiom and states the gap directly:

-- Q3: L2(src_ip)
SELECT SQRT(SUM(c * c))
FROM (SELECT src_ip, COUNT(*) AS c FROM flows WHERE ts >= now() - INTERVAL '1 minute' GROUP BY src_ip);

The SQL frontend does not yet recognize the Q2 and Q3 forms as Entropy and L2. TODO: add this recognition to the per-language frontends.

#509 §Pass 1 (local candidate generation) says Pass 1 "identifies its computation semantics, applies rewrite rules, and generates every candidate", and §Assumptions 1 says the rewrite rules are given by developers. Without a rule that names the computation L2(src_ip), none of the L2(x) candidates (exact L2, norm summary, UnivMon) and none of the Example 2 sharing can start.

Before this PR, the lowered SQL is only relational:

Project [SQRT(col0) AS norm]
└── Aggregate Reduce(none) [Sum(col: product)]
    └── Project [c * c]
        └── Aggregate Reduce(by src_ip) [Count]
            └── Filter / Scan flows

No node carries AggIntent::FrequencyL2, so SemanticEquivalentRewriteStrategy offers nothing. In addition, the native expression compiler rejects the SQL function name sqrt (only promql_sqrt is known), so the original exact SQL could not execute natively either.

Scope. This PR adds one conservative Pass 1 rewrite rule for the floating-point form of the L2 idiom, and native SQL sqrt. It does not recognize the uncast integer form above, which is exactly #509's Q3 text: SQL Int64 c * c can overflow, and that behavior cannot be dropped. It also leaves out the Q2 entropy idiom (#564), the Pass 2 sharing rule, and UnivMon L2 accuracy.

Proposed method

The rule runs in logical ASAP-aware optimization, Pass 1, as part of the existing SemanticEquivalentRewriteStrategy. It is implemented in asap-aware-mapping over the resolved operator graph, not inside the SQL frontend.

Match (frequency_l2_rewrite(root)), each step must hold or the rule returns None:

1. root is a Project with exactly one output.
2. Follow only Project nodes downward, substituting column references by the projected expressions (positional lineage). Only Column, Float64 Cast (try_cast: false), Arithmetic and FunctionCall are substituted. Crossing a Filter or Limit here is not allowed, since that would change the population.
3. The expanded expression is sqrt(arg) (case-insensitive) with one argument, and arg (after removing Float64 casts) is column 0 of the next node.
4. That node is Aggregate { reduction: Reduce(keys), having: None } with no grouping keys, not without, no measure filters, and exactly one measure Sum { col: Some(col) }.
5. Expanding col through projections gives left * right with ExprSemantics::Sql, and its type is already Float64. An Int64 product is refused.
6. The next node is Aggregate { reduction: Reduce(keys), having: None } with exactly one key, not without, no measure filters, and exactly one measure Count { accuracy }. Both left and right (after removing Float64 casts) are column 1, i.e. the count.
7. The grouping key field in the aggregate's child is non-nullable and is Bool, Int64 or Utf8. A nullable key is refused because COUNT(*) GROUP BY makes a real NULL group, while the frequency intent skips NULL.

Rewrite. Build

Project [CASE WHEN col0 = 0.0 THEN CAST(NULL AS DOUBLE) ELSE col0 END AS <original name>]  (same qualifier)
└── Aggregate Reduce(none) [FrequencyL2 { col: Some(key), accuracy: <Count's accuracy> }] AS frequency_l2
    └── <the grouped aggregate's original child>   (filters and scan unchanged)

L2 is positive for any non-empty unit-count population, so 0.0 only happens for an empty one. The CASE restores SQL's NULL for SUM over no groups without another count. The rewrite is returned only if its schema equals the original root's schema.

Candidate. SemanticEquivalentRewriteStrategy::matches and replacements call the rule first. The replacement has provenance: LogicalRewrite. The original exact SQL DAG stays in the search as its own candidate.

Native SQL sqrt. The expression compiler's validate accepts sqrt (case-insensitive) with one Int64, Float64 or Null argument. evaluate returns Null for Null, and Float64(x.sqrt()) for Float64 or Int64 input. A negative input gives NaN, as f64::sqrt does.

Key code interfaces

crates/asap-aware-mapping/src/frequency_rewrite.rs (new private module)

pub(super) fn frequency_l2_rewrite(root: &Rc<OperatorNode>) -> Option<Rc<OperatorNode>>;

// helpers
fn expand(expr: ScalarExpr, node: Rc<OperatorNode>) -> Option<(ScalarExpr, Rc<OperatorNode>)>;
fn substitute(expr: &ScalarExpr, cols: &[ProjectItem]) -> Option<ScalarExpr>;
fn uncast(expr: &ScalarExpr) -> &ScalarExpr;

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

impl ReplacementStrategy for SemanticEquivalentRewriteStrategy {
    fn matches(&self, target: &TargetSubDAG<'_>) -> bool;           // + frequency_l2_rewrite(..).is_some()
    fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec<ReplacementSubDAG>;
}

// replacement produced by the new rule
ReplacementSubDAG {
    strategy: "SemanticEquivalentRewriteStrategy",
    replacement: Replacement::SubDAG(rewritten),
    provenance: ReplacementProvenance::LogicalRewrite,
    rationale: "recognize a floating SQL frequency L2 product while preserving empty-input NULL and the original exact candidate".into(),
}

Intent produced (existing, crates/types/src/pre_asap/agg_intent.rs):

AggIntent::FrequencyL2 { col: Option<C>, accuracy: AccuracyTarget }

Native expression support in crates/asap-physical-operators/src/expressions/planner.rs is inside the private evaluate / validate functions; no public signature changes.

Usage (crates/frontend-sql/tests/frequency_l2.rs):

let root = lower_sql(sql, &catalog, AccuracyTarget::Exact).await?;
let replacements = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root));
// one Replacement::SubDAG(node) whose subtree contains AggIntent::FrequencyL2

Fields

frequency_l2_rewrite

Name	Type	Meaning
root	&Rc<OperatorNode>	Root of the target sub-DAG; must be the single-output outer Project.
return	Option<Rc<OperatorNode>>	Some(rewritten) with a schema equal to root.schema, or None if any match step fails.

Helpers

Name	Meaning
expand(expr, node)	Walks down Project nodes from node, substituting expr's columns. Returns the expanded expression and the first non-Project node.
substitute(expr, cols)	Replaces Column(i) with cols[i].expr. Supports Float64 non-try Cast, Arithmetic, FunctionCall; anything else → None.
uncast(expr)	Strips nested Cast { to: Float64, try_cast: false }.

FrequencyL2 as built by the rule

Field	Value
col	Some(key): the single grouping key index of the inner aggregate, relative to its child.
accuracy	The inner Count { accuracy } target, unchanged. No default is invented.

Rewritten nodes

Node	Field	Value
Aggregate	reduction	Reduce(GroupKeys::none())
	measures / output_names	[FrequencyL2 {..}] / ["frequency_l2"]
	filters / having	[] / None
	child	The inner grouped aggregate's child, so WHERE predicates are kept.
Project	alias	The original root's first field name (e.g. norm).
	expr	CASE WHEN col0 = 0.0 THEN CAST(NULL AS Float64) ELSE col0 END, ExprSemantics::Sql.
	qualifier	The original root's qualifier.

ReplacementSubDAG: strategy names the strategy; replacement is the new sub-DAG; provenance = LogicalRewrite marks a logical rewrite, not a summary realization; rationale is the text above.

SQL sqrt: one argument of type Int64, Float64 or Null; result Float64 or Null.

Examples

End to end (crates/integration-tests/tests/sql_frequency_l2.rs::sql_l2_original_and_rewrite_execute_equivalently).

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
-- flows(src_ip: Utf8 non-null, keep: Bool)

The rule returns a Project over FrequencyL2(src_ip) over the filtered scan. Both the original and the rewrite are compiled to the wire format and executed against a raw in-memory connector:

Input rows (src_ip, keep)	Original SQL	Rewrite
none	NULL	NULL
(discard, false)	NULL	NULL
(a,true),(a,true),(b,true),(discard,false)	√5	√5

Recognized vs declined (crates/frontend-sql/tests/frequency_l2.rs)

SQL inner/outer shape	src_ip nullable	Result
SUM(CAST(c AS DOUBLE)*CAST(c AS DOUBLE)), COUNT(*) AS c … WHERE keep	no	FrequencyL2 candidate; schema equal to original
SUM(c * c), CAST(COUNT(*) AS DOUBLE) AS c	no	FrequencyL2 candidate
SUM(c*c), COUNT(*) AS c (Int64 product, #509 Q3 text)	no	declined (overflow)
SUM(CAST(c AS DOUBLE)*CAST(c AS DOUBLE))	yes	declined (NULL group)
… GROUP BY src_ip HAVING COUNT(*) > 1	no	declined (HAVING)
CAST(SUM(CAST(keep AS BIGINT)) AS DOUBLE) AS c	no	declined (not a unit count)

Other tests in the same file:

• frequency_l2_preserves_accuracy_target: with EpsilonDelta { epsilon: 0.01, delta: 0.01 }, the new intent carries that same target.
• default_search_keeps_exact_sql_and_frequency_alternatives: search_workload_with_targets with default_strategies() yields an inventory that contains both a candidate with FrequencyL2 and one without.

SQL sqrt (crates/asap-physical-operators/tests/physical_semantics.rs::sql_sqrt_executes_numeric_and_null_arguments): sqrt(Int64 9) = 3.0, sqrt(Float64 2.25) = 1.5, sqrt(NULL) = NULL, sqrt(-1.0) is NaN.

Out of scope

• The uncast Int64 c * c form from docs: propose workload-wide planning, summary sharing, and materialization #509 Q3.
• Entropy recognition (feat(planner): recognize normalized SQL entropy in nats #564), sharing partitions, UnivMon L2 accuracy certification.
• Integer products, nullable-key groups, filtered measures, HAVING and other population-changing shapes remain ordinary SQL.

The status note docs/develop_docs/planner-layering-status.md gains §"SQL L2 follow-up acceptance" describing the same limits.

Stack and validation

Stacked on #559 · Next: #563 · Reference/tracker: #528

Validation: recognition regression fails on the parent; four frontend tests cover recognition, refusal boundaries, accuracy and inventory retention; end-to-end comparison covers filtered data and empty input; native scalar tests; existing rewrite tests; formatting and affected all-target Clippy with warnings denied.

🤖 Generated with Claude Code

[zzylol]

Parked as draft: PR priorities changed (see #528). Order is now (A) finish #511 operator sharing, (B) the #572 crate/module reorganization, (C) #509 end-to-end stages. This PR sits on the old stack/528-legacy-physical-base chain, and Phase B moves the files it touches. Its content will be re-scoped onto the new layout in Phase C.

🤖 Generated with Claude Code

[zzylol]

Ported onto the current stack in #598