feat(runtime): execute exact frequency L2 and entropy

[zzylol]

Problem: the exact L2 and entropy candidates of #509 Example 2 cannot execute

#509 §Pass 1 (local candidate generation) lists an exact candidate next to the summary candidates for both frequency statistics:

Original computation	Local candidates (#509 §Pass 1 table)
Entropy(x)	Exact entropy, a specialized entropy summary, UnivMon
L2(x)	Exact L2 norm, a specialized norm summary, UnivMon

#509 §Example 2 uses these for Q2 and Q3: "each gets its local candidates from the Pass 1 table: exact, a specialized summary, or UnivMon". The independent candidates "are kept as well" after Pass 2, so selection can always fall back to the exact plan. #509 §Physical operator implementation must then turn every node, including the exact one, into a physical operator.

Before this PR, the IR already has AggIntent::FrequencyL2 and AggIntent::FrequencyEntropy, but no exact physical implementation exists:

Aggregate { reduction: Reduce(none), measures: [FrequencyL2 { col: Some(0), accuracy: Exact }] }
  native binding (physical_planner::bind_operation)
    → Err("aggregate intent has no native implementation")
  raw analytical costing (query_physical_lowering, supports_hash_aggregate)
    → Err(AnalyticalCostError::UnsupportedQueryOperator)

The only way to answer these queries is a manually built UnivMon state (#557). That is an approximation, not the exact fallback #509 assumes.

Scope. This PR supplies the exact physical operator for the L2(x) and Entropy(x) rows of the #509 §Pass 1 table, and makes the raw cost path accept them. It does not add SQL recognition of the Q2/Q3 idioms (#509 §Example 2 TODO), Pass 2 sharing, or UnivMon accuracy. It also adds a developer status note that audits #557 against #509 and lists the remaining scopes in order.

Proposed method

Physical runtime and raw costing only. Logical planning is unchanged.

1. Native binding. bind_operation maps AggIntent::FrequencyL2 { col, .. } to Reduction::FrequencyL2(column) and AggIntent::FrequencyEntropy { col, .. } to Reduction::FrequencyEntropy(column). As for other intents, col: None resolves to the sample-value column.
2. Construction check. Operator::aggregate accepts these reductions only over a plain Bool, Int64, Float64 or Utf8 column. The output column is non-nullable Float64.
3. Exact reducer. For each group, reduce_one:
   • skips NULL;
   • rejects non-finite Float64 identities;
   • encodes the value with the type-tagged Value::key() (so -0.0 and 0.0 are one key, and Int64 keys above 2^53 stay distinct);
   • counts each key in a BTreeMap<Vec<u8>, u64> and tracks the total;
   • computes L2 as the running hypot of all counts, or entropy as -Σ p·log2(p) with p = count / total.
     An empty population (no rows or only NULL) returns 0.0.
4. Memory and cancellation. Each new key reserves 64 + size_of::<Vec<u8>>() + key.len() + size_of::<u64>() bytes in a Workspace tied to the RunContext. The reservation fails with Error::MemoryLimit when over budget and is released afterward. Each row and each count passes work.checkpoint(), so long reductions yield and can be cancelled. reduce_one now receives the RunContext for this.
5. Raw costing. supports_hash_aggregate lists FrequencyL2 and FrequencyEntropy, so a raw plan with these intents lowers to PhysicalOperator::HashAggregate instead of failing.

Entropy is in bits, matching the AggIntent::FrequencyEntropy doc ("Shannon entropy in bits").

Key code interfaces

crates/asap-physical-operators/src/operators/aggregate/mod.rs

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
pub enum Reduction {
    Count,
    Sum(usize),
    Avg(usize),
    Min(usize),
    Max(usize),
    /// L2 norm of unit-update frequencies; NULL identities are skipped.
    FrequencyL2(usize),
    /// Shannon entropy in bits; NULL identities are skipped.
    FrequencyEntropy(usize),
    Quantile { column: usize, q: f64 },
}

impl Operator {
    pub fn aggregate(
        input: SchemaRef,
        groups: Vec<usize>,
        measures: Vec<(String, Reduction)>,
    ) -> Result<Self, Error>;
}

Existing IR intents now bound natively (crates/types/src/pre_asap/agg_intent.rs, unchanged):

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

Binding (crates/asap-physical-operators/src/physical_planner/mod.rs) and raw costing (crates/asap-aware-mapping/src/query_physical_lowering.rs):

AggIntent::FrequencyL2 { col, .. } => Reduction::FrequencyL2(column(*col)?),
AggIntent::FrequencyEntropy { col, .. } => Reduction::FrequencyEntropy(column(*col)?),

fn supports_hash_aggregate(reduction: &Reduction, measures: &[AggIntent]) -> bool; // + FrequencyL2, FrequencyEntropy

Usage (tests/physical_semantics.rs::exact_frequency_grouping_and_entropy_bits):

let operator = Operator::aggregate(
    input,                 // (group: Int64, key: Float64 nullable)
    vec![0],
    vec![
        ("l2".into(), Reduction::FrequencyL2(1)),
        ("entropy".into(), Reduction::FrequencyEntropy(1)),
    ],
)?;

Fields

Reduction (new variants)

Variant	Field	Meaning / invariants
FrequencyL2(usize)	column index	Identity column. Must be plain Bool, Int64, Float64 or Utf8. Output: sqrt(Σ count_k²) over distinct non-NULL keys k in the group, as non-nullable Float64. 0.0 for an empty population.
FrequencyEntropy(usize)	column index	Same column rule. Output: -Σ p_k·log2(p_k) in bits, p_k = count_k / total. 0.0 for an empty or single-key population.

Both: NULL identities are skipped; non-finite Float64 identities are an error at execution; each row counts once (unit update).

Operator::aggregate

Parameter	Type	Meaning
input	SchemaRef	Input row schema.
groups	Vec<usize>	Grouping columns. Each group gets its own frequency population. Empty = one global row.
measures	Vec<(String, Reduction)>	Output name and reduction per measure.

AggIntent::FrequencyL2 / FrequencyEntropy (as used by binding)

Field	Type	Use in this PR
col	Option<usize>	Identity column. None resolves to ColumnRef::SampleValue like other intents.
accuracy	AccuracyTarget	Ignored by binding. The exact reducer always returns the exact value.

supports_hash_aggregate

Parameter	Meaning
reduction	Must be Reduction::Reduce(_).
measures	Non-empty; every intent must be in the allowed list, which now includes the two frequency intents.

Examples

End to end through binding (tests/physical_semantics.rs::exact_frequency_intents_execute_typed_keys_and_empty_input). A PostAsapDAGNode with Aggregate { measures: [FrequencyL2 | FrequencyEntropy { col: Some(0), accuracy: Exact }] } is compiled with compile_node and run. Input is two keys, each twice, then one NULL (a,a,b,b,NULL):

Key type	Keys	L2	Entropy (bits)
Utf8	"a", "b"	√8	1
Int64	9_007_199_254_740_992, 9_007_199_254_740_993	√8	1
Bool	false, true	√8	1
Float64	-0.0, 1.0	√8	1

Empty input and NULL-only input both return Float64(0.0) for both intents. Results are checked to 1e-12.

Grouping (exact_frequency_grouping_and_entropy_bits).

group	key values	L2	entropy
1	-0.0, 0.0, 0.0, 1.0 (zero ×3, one ×1)	√10	-0.75·log2(0.75) - 0.25·log2(0.25)
2	2.0, NULL	1	0
3	NULL	0	0

Grouped aggregation over no input rows returns no rows.

Memory (tests/blocking_resources.rs::frequency_dictionary_enforces_memory_budget). 64 distinct Int64 rows under a 12,000-byte budget:

Reduction	Result
Count	succeeds
FrequencyL2(0)	Err(Error::MemoryLimit)
FrequencyEntropy(0)	Err(Error::MemoryLimit)

In every case retained_bytes() is 0 afterward.

Accepted vs rejected

Case	Result
identity Bool / Int64 / Float64 / Utf8	accepted
other identity type	rejected by Operator::aggregate
non-finite Float64 value	error at execution
NULL value	skipped

Status note. docs/develop_docs/planner-layering-status.md (new) audits #557 against #509 per contract (frontends, local alternatives, summary-capability sharing, window composition, materialization, whole-workload selection, deployment inputs, TODOs) and lists seven ordered follow-up scopes. Step 1, exact frequency execution, is this PR.

Out of scope

• SQL recognition of the Q2/Q3 idioms (feat(planner): recognize floating SQL frequency L2 idioms #562 starts this).
• Exact distinct (Cardinality) binding (feat(runtime): execute exact distinct identity counts #563).
• UnivMon accuracy certification. This PR does not claim the proposal's full examples are implemented.

Stack and validation

Stacked on #557 · Next: #562 · Reference/tracker: #528

Validation: native binding regression failed before the implementation and passes afterward; native runtime suite; focused grouping and dictionary-memory tests; formatting; native/runtime and mapping 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