feat(runtime): execute exact distinct identity counts

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

@@ -1203,6 +1203,7 @@ fn supports_hash_aggregate(
                     | AggIntent::PearsonCorr { .. }
                     | AggIntent::Group
                     | AggIntent::CountValues { .. }
+                    | AggIntent::Cardinality { .. }
                     | AggIntent::FrequencyL2 { .. }
                     | AggIntent::FrequencyEntropy { .. }
             )

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

@@ -13,6 +13,17 @@ impl Operator {
         for (name, reduction) in &measures {
             let (t, n) = match reduction {
                 Reduction::Count => (DataType::Int64, false),
+                Reduction::Cardinality(columns) => {
+                    if columns.is_empty() {
+                        return Err(invalid(
+                            "distinct aggregate requires at least one identity column",
+                        ));
+                    }
+                    for column in columns {
+                        plain(&input, *column)?;
+                    }
+                    (DataType::Int64, false)
+                }
                 Reduction::Sum(i) | Reduction::Avg(i) => {
                     let (t, _) = plain(&input, *i)?;
                     if !matches!(t, DataType::Int64 | DataType::Float64) {
@@ -134,6 +145,8 @@ impl Operator {
 #[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(usize),
     Min(usize),
@@ -265,6 +278,27 @@ async fn reduce_one(
             ))
         }
         Reduction::Sum(i) | Reduction::Avg(i) | Reduction::Min(i) | Reduction::Max(i) => *i,
+        Reduction::Cardinality(columns) => {
+            let mut workspace = Workspace::new(context)?;
+            let mut identities = std::collections::BTreeSet::new();
+            for row in rows {
+                work.checkpoint().await?;
+                if columns
+                    .iter()
+                    .any(|column| matches!(row[*column], Value::Null))
+                {
+                    continue;
+                }
+                let key = group_key(row, columns)?;
+                if !identities.contains(&key) {
+                    workspace.grow(key_bytes(&key))?;
+                    identities.insert(key);
+                }
+            }
+            return Ok(Value::Int64(
+                i64::try_from(identities.len()).map_err(|_| invalid("distinct count overflow"))?,
+            ));
+        }
         Reduction::FrequencyL2(column) | Reduction::FrequencyEntropy(column) => {
             let mut workspace = Workspace::new(context)?;
             let mut counts = BTreeMap::<Vec<u8>, u64>::new();

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

@@ -826,6 +826,13 @@ fn bind_operation(node: &PostAsapDAGNode, inputs: &[SchemaRef]) -> Result<Operat
                         };
                         let m = match m {
                             AggIntent::Count { .. } => Reduction::Count,
+                            AggIntent::Cardinality { cols, .. } => {
+                                Reduction::Cardinality(if cols.is_empty() {
+                                    vec![column(None)?]
+                                } else {
+                                    cols.clone()
+                                })
+                            }
                             AggIntent::Sum { col } => Reduction::Sum(column(*col)?),
                             AggIntent::Avg { col } => Reduction::Avg(column(*col)?),
                             AggIntent::FrequencyL2 { col, .. } => {

crates/asap-physical-operators/tests/blocking_resources.rs

@@ -274,6 +274,7 @@ fn frequency_dictionary_enforces_memory_budget() {
         (Reduction::Count, true),
         (Reduction::FrequencyL2(0), false),
         (Reduction::FrequencyEntropy(0), false),
+        (Reduction::Cardinality(vec![0]), false),
     ] {
         let run = context(12_000);
         let inputs = sources.execute(&[0], run.clone()).unwrap();

crates/asap-physical-operators/tests/physical_semantics.rs

@@ -861,3 +861,90 @@ fn sql_sqrt_executes_numeric_and_null_arguments() {
         matches!(compiled.evaluate(&[Value::Float64(-1.0)]).unwrap(), Value::Float64(v) if v.is_nan())
     );
 }
+
+// Exact distinct binding preserves typed tuples, skips NULLs and returns zero on empty input.
+#[test]
+fn exact_cardinality_binds_and_executes_typed_tuples() {
+    use asap_physical_operators::physical_planner::compile_node;
+    use planner_types::{
+        post_asap::ExecutionDataState,
+        pre_asap::{AggIntent, GroupKeys, Reduction as PlanReduction},
+        types::AccuracyTarget,
+    };
+    let input = schema(&[
+        ("key", DataType::Int64, true),
+        ("tag", DataType::Utf8, true),
+    ]);
+    for (cols, expected) in [(vec![0], 2), (vec![0, 1], 3)] {
+        let node = PostAsapDAGNode {
+            id: PostAsapNodeId(1),
+            payload: PostAsapOperatorPayload::Relational {
+                operator: ValueOperation::Aggregate {
+                    reduction: PlanReduction::Reduce(GroupKeys::none()),
+                    measures: vec![AggIntent::Cardinality {
+                        cols,
+                        accuracy: AccuracyTarget::Exact,
+                    }],
+                    output_names: vec!["distinct".into()],
+                    filters: vec![],
+                    having: None,
+                },
+            },
+            output_state: ExecutionDataState::QUERY_ROWS,
+            output_schema: (*schema(&[("distinct", DataType::Int64, false)])).clone(),
+            guarantee: None,
+        };
+        let operator =
+            compile_node(&node, std::slice::from_ref(&input)).expect("exact distinct intent binds");
+        let rows = vec![
+            vec![Value::Int64(9_007_199_254_740_992), Value::Utf8("a".into())],
+            vec![Value::Int64(9_007_199_254_740_992), Value::Utf8("a".into())],
+            vec![Value::Int64(9_007_199_254_740_992), Value::Utf8("b".into())],
+            vec![Value::Int64(9_007_199_254_740_993), Value::Utf8("a".into())],
+            vec![Value::Null, Value::Utf8("c".into())],
+        ];
+        let result = unary(input.clone(), vec![rows], operator.clone());
+        assert!(matches!(result[0][0], Value::Int64(v) if v == expected));
+        for rows in [vec![], vec![vec![Value::Null, Value::Null]]] {
+            let result = unary(input.clone(), vec![rows], operator.clone());
+            assert!(matches!(result[0][0], Value::Int64(0)));
+        }
+    }
+}
+
+// Distinct uses grouped equality: signed zero and NaN payloads each form one identity.
+#[test]
+fn exact_cardinality_grouping_normalizes_float_identities() {
+    let input = schema(&[
+        ("group", DataType::Int64, false),
+        ("key", DataType::Float64, true),
+    ]);
+    let operator = Operator::aggregate(
+        input.clone(),
+        vec![0],
+        vec![("distinct".into(), Reduction::Cardinality(vec![1]))],
+    )
+    .unwrap();
+    let result = unary(
+        input,
+        vec![vec![
+            vec![Value::Int64(1), Value::Float64(0.0)],
+            vec![Value::Int64(1), Value::Float64(-0.0)],
+            vec![Value::Int64(1), Value::Float64(f64::NAN)],
+            vec![
+                Value::Int64(1),
+                Value::Float64(f64::from_bits(f64::NAN.to_bits() + 1)),
+            ],
+            vec![Value::Int64(2), Value::Null],
+        ]],
+        operator,
+    );
+    assert!(matches!(
+        result[0].as_slice(),
+        [Value::Int64(1), Value::Int64(2)]
+    ));
+    assert!(matches!(
+        result[1].as_slice(),
+        [Value::Int64(2), Value::Int64(0)]
+    ));
+}

crates/integration-tests/tests/physical_common/mod.rs

@@ -53,3 +53,67 @@ pub fn compile_post_asap_dag(
     )?;
     Ok(asap_types::ir::export::compile_post_asap_dag(&root)?)
 }
+
+// Execute a selected relational DAG against real raw connectors, including scan predicates.
+#[allow(dead_code)]
+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>> {
+    use asap_physical_operators::{
+        physical_planner::bind_with_data_sources,
+        runtime::{Limits, RunContext},
+        sources::{DataSources, MemorySource},
+    };
+    use asap_types::ir::export::{NonASAPOpKind, PostAsapOperatorPayload};
+    use futures::{executor::block_on, StreamExt};
+    use std::sync::Arc;
+    let wire = 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
+}

crates/integration-tests/tests/sql_cardinality.rs

@@ -0,0 +1,46 @@
+//! The design's SQL distinct query executes its exact native fallback.
+mod physical_common;
+use asap_frontend_sql::{lower_sql, SqlCatalog};
+use asap_physical_operators::values::Value;
+use asap_types::{
+    pre_asap::{DataType, Field, Schema},
+    types::AccuracyTarget,
+};
+
+// COUNT(DISTINCT src_ip) skips NULL and preserves WHERE, Utf8 identities and zero on no input.
+#[tokio::test]
+async fn sql_distinct_executes_through_raw_scan_and_native_binding() {
+    let catalog = SqlCatalog::new().with_table(
+        "flows",
+        Schema::new(vec![
+            Field::plain("src_ip", DataType::Utf8, true),
+            Field::plain("keep", DataType::Bool, false),
+        ]),
+    );
+    let root = lower_sql(
+        "SELECT COUNT(DISTINCT src_ip) AS sources FROM flows WHERE keep",
+        &catalog,
+        AccuracyTarget::Exact,
+    )
+    .await
+    .unwrap();
+    for (rows, expected) in [
+        (vec![], 0),
+        (vec![vec![Value::Null, Value::Bool(true)]], 0),
+        (
+            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)],
+                vec![Value::Null, Value::Bool(true)],
+            ],
+            2,
+        ),
+    ] {
+        let result = physical_common::execute_raw_rows(&root, rows);
+        assert!(
+            matches!(result.as_slice(), [row] if matches!(row.as_slice(), [Value::Int64(v)] if *v == expected))
+        );
+    }
+}

crates/integration-tests/tests/sql_frequency_l2.rs

@@ -5,76 +5,13 @@ use asap_aware_mapping::{
     SemanticEquivalentRewriteStrategy,
 };
 use asap_frontend_sql::{lower_sql, SqlCatalog};
-use asap_physical_operators::{
-    runtime::Scope,
-    values::{Batch, Value},
-};
+use asap_physical_operators::values::Value;
 use asap_types::{
-    ir::{
-        export::{NonASAPOpKind, PostAsapOperatorPayload},
-        NonASAPOp, OperatorNode,
-    },
+    ir::NonASAPOp,
     pre_asap::{DataType, Field, Schema},
     types::AccuracyTarget,
 };
-use std::{collections::BTreeMap, rc::Rc, sync::Arc};
 
-fn run(root: &Rc<OperatorNode>, rows: Vec<Vec<Value>>) -> Vec<Vec<Value>> {
-    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() {
@@ -104,8 +41,8 @@ async fn sql_l2_original_and_rewrite_execute_equivalently() {
             vec![Value::Utf8("discard".into()), Value::Bool(false)],
         ],
     ] {
-        let original = run(&root, rows.clone());
-        let actual = run(rewritten, rows);
+        let original = physical_common::execute_raw_rows(&root, rows.clone());
+        let actual = physical_common::execute_raw_rows(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)