feat(runtime): execute exact SQL entropy fallback

[zzylol]

Problem: the exact candidate for #509 Example 2 Q2 cannot run natively

#509 §Pass 1: Local candidate generation gives Entropy(x) three local candidates: "Exact entropy, a specialized entropy summary, UnivMon". #509 §Stages and their decisions requires each stage to output "every semantically equivalent and legal candidate", and §3 Plan selection rejects a candidate "that needs a capability the deployment lacks". So the original exact Q2 graph must stay a real, executable candidate next to the rewrite from #564.

#564 keeps the original Q2 graph as an alternative. But native binding cannot run it. Q2 (#509 §Example 2) is:

SELECT -SUM(p * LN(p))
FROM (
  SELECT COUNT(*) * 1.0 / SUM(COUNT(*)) OVER () AS p
  FROM flows
  WHERE ts >= now() - INTERVAL '1 minute'
  GROUP BY src_ip
);

Before this PR, two pieces of this graph have no native implementation:

Piece	IR	Before
SUM(COUNT(*)) OVER ()	NonASAPOp::SQLWindowFunc { func: Sum, partition_by: [], order_by: [], frame: UNBOUNDED PRECEDING … UNBOUNDED FOLLOWING }	bind_operation falls to value operation has no native implementation
LN(p)	ScalarExpr::FunctionCall { name: "ln", .. }	expression validate rejects it. Only SQL sqrt, the promql_-prefixed functions (e.g. promql_ln) and map functions are accepted

So in native execution only the rewritten frequency graph from #564 runs, and there is no way to check that it gives the same answer as the SQL it replaced.

Scope covered here. Native execution of the exact Q2 graph: SQL LN, and SUM(column) OVER () over the complete, unordered relation.

Left out. Partitioned, ordered and finite window frames (still refused). Raw analytical cost lowering for this window. This SQL window is not the streaming summary window of #509 §Pass 2 (sliding / tumbling / EH); that is #566.

Proposed method

All changes are in the native physical layer (asap-physical-operators), stage 2 physical operator implementation and stage 4 execution.

1. New physical operator SQLWindowSum. Operator::sql_window_sum(input, column, name) checks that column is a plain Int64 or Float64 field. The output schema is the input schema plus one nullable field name of the same type. Input metadata is kept because the input schema is cloned.
2. Execution. It runs in the existing blocking aggregate executor (aggregate::execute). It collects all input rows, computes one total with the existing reduce_one(.., Reduction::Sum(column), ..), and appends that total to every row. Row order and row count are unchanged. Empty or all-NULL input gives a NULL total (SUM semantics). Each output row is charged to Workspace, so a memory limit fails cleanly with Error::MemoryLimit. Cooperative checkpoints are taken per row.
3. Binding. bind_operation maps NonASAPOpKind::SQLWindowFunc to SQLWindowSum only when: func is Sum; args is exactly one column; partition_by is not without and has no keys; order_by is empty; the frame start is Preceding(Null) and the end is Following(Null) (both unbounded). Anything else returns native SQL window SUM requires the complete unordered relation.
4. Recovery. UncheckedOperator → Operator rebuilds SQLWindowSum, taking the output name from the last output field. So a serialized operator round-trips through the same constructor checks.
5. LN. The expression evaluator handles ln with the same path as sqrt: NULL → NULL, Int64/Float64 → Float64(value.ln()), anything else is SQL math function requires a numeric argument. validate accepts ln with one Int64/Float64/Null argument.

Key code interfaces

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

impl Operator {
    /// A SQL SUM window over the complete, unordered input relation.
    pub fn sql_window_sum(input: SchemaRef, column: usize, name: String) -> Result<Self, Error>;
}

crates/asap-physical-operators/src/operators/mod.rs (private Kind, display name "SQLWindowSum"):

enum Kind {
    // …
    SQLWindowSum { column: usize },
    // …
}

crates/asap-physical-operators/src/physical_planner/mod.rs, new arm in bind_operation:

NonASAPOpKind::SQLWindowFunc {
    func: WindowFuncKind::Sum,
    args,          // must be [WireScalarExpr::Column(column)]
    partition_by,  // must be empty and not `without`
    order_by,      // must be empty
    frame: Some(frame), // UNBOUNDED PRECEDING .. UNBOUNDED FOLLOWING
    output_name,
} => Operator::sql_window_sum(input.clone(), *column, output_name.clone()),

crates/asap-physical-operators/src/expressions/planner.rs: evaluate and validate accept FunctionCall { name: "ln", args: [x] } (case-insensitive).

Usage (from crates/asap-physical-operators/tests/physical_semantics.rs):

let input = schema(&[("v", DataType::Int64, true)]);
let operator = Operator::sql_window_sum(input.clone(), 0, "total".into())?;

Fields

Operator::sql_window_sum

Item	Type	Meaning / rules
input	SchemaRef	Input relation schema. Copied into the output.
column	usize	Index of the summed field. Must be a plain Int64 or Float64 field, else SQL window SUM requires a numeric column.
name	String	Name of the appended total field. Set by binding from output_name, or by recovery from the last output field.
return	Result<Operator, Error>	Operator with inputs = [input] and output = input fields + (name, same dtype, nullable = true).

Kind::SQLWindowSum

Field	Type	Meaning
column	usize	Same as above. The only state; the total is computed per execution.

SQLWindowFunc binding conditions

Wire field	Accepted value	Why
func	WindowFuncKind::Sum	Only SUM is implemented.
args	exactly [Column(c)]	Argument must be projected to a column first.
partition_by	not without, no keys	One total for the whole relation.
order_by	empty	No running total.
frame	Some, start Preceding(Scalar(Null)), end Following(Scalar(Null))	Complete frame; a finite frame is a different value.
output_name	any	Becomes name.

ln scalar function

Input	Output
NULL	NULL
Int64(v) / Float64(v)	Float64(v.ln())
other	error SQL math function requires a numeric argument

Examples

End-to-end: original Q2 vs. #564 rewrite (crates/integration-tests/tests/sql_frequency_entropy.rs). Table flows(src_ip Utf8, keep Bool). Query:

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

The test executes both the original lowered graph and the rewritten graph on the same rows (each case adds one ("discard", false) row) and requires equal results:

Kept src_ip values	Original SQL	Rewrite
none	NULL	NULL
a, a	-0.0	-0.0
a, a, b, b	ln 2	ln 2
a, a, a, b	-0.75·ln 0.75 − 0.25·ln 0.25	same, within 1e-12

For a, a, b, b the original graph runs as: group count {a: 2, b: 2} → SQLWindowSum appends total = 4 to both rows → p = 0.5 → -SUM(0.5·ln 0.5 · 2) = ln 2.

Operator semantics (physical_semantics.rs, complete_sql_sum_window_preserves_rows_and_nulls, run after a serde round trip):

Input v (nullable Int64)	Output rows
no rows	no rows
[NULL]	[NULL, NULL]
[1], [NULL], [3]	[1, 4], [NULL, 4], [3, 4]

Binding: accepted vs refused (native_sql_sum_window_rejects_other_frames):

SQL window	Result
SUM(x) OVER ()	binds to SQLWindowSum
SUM(src_ip) OVER (PARTITION BY src_ip)	refused
SUM(src_ip) OVER (ORDER BY src_ip)	refused
SUM(src_ip) OVER (ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)	refused

Other tests:

• blocking_resources.rs, complete_window_sum_enforces_workspace_budget: 64 rows with a 12,000-byte budget fail with Error::MemoryLimit, and retained bytes return to 0 after drop.
• physical_semantics.rs, sql_ln_executes_numeric_and_null_arguments: ln(2) on Int64 and Float64 gives LN_2; ln(NULL) gives NULL.

Out of scope

• Partitioned, ordered and finite frames; other window functions.
• Raw analytical cost lowering for the unordered SUM window. Native execution does not supply cost evidence.
• Streaming summary-window composition (docs: propose workload-wide planning, summary sharing, and materialization #509 §Pass 2 window-composition rule): Generate exact window composition and materialization candidates #566.
• Docs: docs/develop_docs/planner-layering-status.md gains a "Native entropy fallback follow-up acceptance" section, and docs/develop_docs/README.md now links it.

Stack and validation

Stacked on #564 (stack/509-19-sql-frequency-entropy). Head: stack/509-20-sql-entropy-fallback. Next: #566.

Validation (from the current PR body):

• The original-SQL regression fails on the parent and passes afterward.
• Semantic, resource, recovery and frame-refusal tests pass.
• The final stacked tip passes all 1,630 workspace tests/doctests (two existing ignores), formatting, and workspace/all-target/all-feature 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 #599