feat(runtime): execute exact distinct identity counts

[zzylol]

Problem: the exact Distinct(x) candidate of #509 Example 2 cannot execute

#509 §Pass 1 (local candidate generation) gives Distinct(x) an exact candidate:

Original computation	Local candidates (#509 §Pass 1 table)
Distinct(x)	Exact distinct, a specialized distinct summary, UnivMon

#509 §Example 2 Q1 is that computation:

-- Q1: Distinct(src_ip)
SELECT COUNT(DISTINCT src_ip) FROM flows WHERE ts >= now() - INTERVAL '1 minute';

Pass 2 keeps the independent candidates next to the shared UnivMon, and #509 §Physical operator implementation must convert every node to a physical operator. So the exact distinct plan must run.

Before this PR, the SQL frontend already lowers Q1 to AggIntent::Cardinality, but the two physical paths refuse it:

Aggregate Reduce(none) [Cardinality { cols: [src_ip], 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)

#559 added the exact L2 and entropy reducers. Distinct is the third Example 2 statistic still without one.

Scope. This PR covers the Distinct(x) exact row of the #509 §Pass 1 table: native execution of Cardinality and raw cost eligibility. It does not change sketch sizing, accuracy certificates, filtered-measure support, entropy recognition or window planning.

Proposed method

Physical runtime and raw costing only.

1. Native binding. bind_operation maps AggIntent::Cardinality { cols, .. } to Reduction::Cardinality(cols). If cols is empty, it uses the sample-value column (ColumnRef::SampleValue), the same convention as other intents with no column.
2. Construction check. Operator::aggregate rejects an empty column list ("distinct aggregate requires at least one identity column"). Each listed column must be plain. The output is non-nullable Int64.
3. Exact reducer. For each group, reduce_one:
   • skips a row if any identity column is NULL (SQL COUNT(DISTINCT a, b) semantics);
   • builds the key with the existing group_key(row, columns), which yields one type-tagged byte string per column (Vec<Vec<u8>>), so tuple boundaries are kept, Int64 values above 2^53 stay distinct, and -0.0/0.0 and NaN payloads each collapse to one identity;
   • inserts new keys into a BTreeSet;
   • returns Int64(len). Empty input gives Int64(0).
4. Memory and cancellation. Each new key reserves key_bytes(&key) in a Workspace on the RunContext. Over budget fails with Error::MemoryLimit, and the reservation is released afterward. Every row passes work.checkpoint().
5. Raw costing. supports_hash_aggregate now lists Cardinality, so raw plans lower to PhysicalOperator::HashAggregate.
6. Test helper. The raw-connector runner used by feat(planner): recognize floating SQL frequency L2 idioms #562's L2 test moves to crates/integration-tests/tests/physical_common/mod.rs as execute_raw_rows, shared by the L2 and distinct SQL tests.

Key code interfaces

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

#[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
pub enum Reduction {
    Count,
    /// Exact distinct tuple count; a tuple with any NULL component is skipped.
    Cardinality(Vec<usize>),
    Sum(usize),
    // … Avg, Min, Max, FrequencyL2, FrequencyEntropy, Quantile unchanged
}

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

Cardinality { cols: Vec<C>, accuracy: AccuracyTarget },

Binding (crates/asap-physical-operators/src/physical_planner/mod.rs):

AggIntent::Cardinality { cols, .. } => Reduction::Cardinality(if cols.is_empty() {
    vec![column(None)?]          // sample-value column
} else {
    cols.clone()
}),

Raw costing (crates/asap-aware-mapping/src/query_physical_lowering.rs): supports_hash_aggregate accepts AggIntent::Cardinality { .. }.

Test helper (crates/integration-tests/tests/physical_common/mod.rs):

pub fn execute_raw_rows(
    root: &std::rc::Rc<asap_types::ir::OperatorNode>,
    rows: Vec<Vec<asap_physical_operators::values::Value>>,
) -> Vec<Vec<asap_physical_operators::values::Value>>;

Fields

Reduction::Cardinality(Vec<usize>)

Field	Meaning / invariants
Vec<usize>	Ordered identity columns. Non-empty; each must be a plain column. One column = distinct values; several = distinct tuples. A row with any NULL in these columns is skipped. Output: non-nullable Int64 count per group, 0 for an empty population.

AggIntent::Cardinality (as used by binding)

Field	Type	Use in this PR
cols	Vec<usize>	Identity columns, copied to Reduction::Cardinality. Empty → [SampleValue column].
accuracy	AccuracyTarget	Ignored by binding; the exact reducer always returns the exact count.

execute_raw_rows

Parameter	Meaning
root	Logical root. Compiled with compile_post_asap_dag; the first Scan node gives the source name and schema.
rows	Rows registered as a MemorySource for that source.
return	All output rows. The helper also asserts retained_bytes() == 0 after execution.

Examples

End to end (crates/integration-tests/tests/sql_cardinality.rs::sql_distinct_executes_through_raw_scan_and_native_binding).

SELECT COUNT(DISTINCT src_ip) AS sources FROM flows WHERE keep
-- flows(src_ip: Utf8 nullable, keep: Bool)

lower_sql → compile_post_asap_dag → bind_with_data_sources with a raw in-memory connector → execute:

Input rows (src_ip, keep)	Result
none	0
(NULL, true)	0
(a,true),(a,true),(b,true),(discard,false),(NULL,true)	2

The WHERE keep predicate removes discard; NULL is skipped.

Typed tuples (crates/asap-physical-operators/tests/physical_semantics.rs::exact_cardinality_binds_and_executes_typed_tuples). A PostAsapDAGNode with Cardinality { cols, accuracy: Exact } is compiled with compile_node. Rows (key: Int64, tag: Utf8):

(2^53,   "a")
(2^53,   "a")
(2^53,   "b")
(2^53+1, "a")
(NULL,   "c")

cols	Result	Why
[0]	2	2^53 and 2^53+1 are distinct; NULL row skipped
[0, 1]	3	(2^53,a), (2^53,b), (2^53+1,a); NULL-containing tuple skipped

For both, empty input and an all-NULL row return Int64(0).

Grouping and float identities (exact_cardinality_grouping_normalizes_float_identities).

group	key values	distinct
1	0.0, -0.0, NaN, NaN with a different payload	2
2	NULL	0

Memory (tests/blocking_resources.rs::frequency_dictionary_enforces_memory_budget). With 64 distinct Int64 rows under a 12,000-byte budget, Cardinality(vec![0]) fails with Error::MemoryLimit (like FrequencyL2/FrequencyEntropy; Count succeeds), and retained_bytes() returns to 0.

Accepted vs rejected

Case	Result
one or more plain identity columns	accepted
empty column list in Reduction::Cardinality	rejected by Operator::aggregate
empty cols in AggIntent::Cardinality	bound to the sample-value column
row with any NULL identity component	skipped
filtered aggregate measure (filters with Some)	still rejected by native binding (existing scope)

The status note docs/develop_docs/planner-layering-status.md gains §"Exact distinct follow-up acceptance".

Out of scope

• Sketch sizing and accuracy certificates for distinct.
• Filtered-measure support, entropy recognition, window planning.

Stack and validation

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

Validation: binding regression fails on the parent and passes after implementation; native semantic/resource tests; SQL distinct and L2 wire execution; raw physical-lowering 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 #597