feat: add native physical operators and shared DAG runtime

crates/asap-physical-operators/Cargo.toml

@@ -4,9 +4,11 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+futures = "0.3"
 planner-types = { package = "asap-types", path = "../types" }
 asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "5f03ccbd798ed5fec62bdd839bcb331123cab369" }
 serde = { version = "1", features = ["derive", "rc"] }
+serde_json = "1"
 tracing = "0.1"
 thiserror = "2"
 

crates/asap-physical-operators/src/error.rs

@@ -1,3 +1,4 @@
+use crate::plan::NodeId;
 #[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
 pub enum Error {
     #[error("invalid DAG: {0}")]
@@ -6,7 +7,7 @@ pub enum Error {
     Operator(String),
     #[error("node {node} ({operation}) failed: {source}")]
     AtNode {
-        node: u64,
+        node: NodeId,
         operation: String,
         source: Box<Error>,
     },

crates/asap-physical-operators/src/expressions/arithmetic.rs

@@ -0,0 +1,63 @@
+//! Float64 arithmetic shared by ASAP execution engines.
+//! Preserve IEEE non-finite results; callers own their output policies.
+
+pub fn evaluate_float64_arithmetic(
+    operator: &planner_types::pre_asap::ArithmeticOpKind,
+    left: f64,
+    right: f64,
+) -> f64 {
+    use planner_types::pre_asap::ArithmeticOpKind::*;
+    match operator {
+        Add => left + right,
+        Sub => left - right,
+        Mul => left * right,
+        Div => left / right,
+        Mod => left % right,
+        Pow => left.powf(right),
+        Atan2 => left.atan2(right),
+    }
+}
+
+/// Execute the Planner binary contract after a deployment has resolved matching rows.
+pub fn evaluate_binary(
+    operator: &planner_types::post_asap::BinaryOperator,
+    left: f64,
+    right: f64,
+) -> Result<crate::values::Value, crate::Error> {
+    use crate::{values::Value, Error};
+    use planner_types::pre_asap::{ArithmeticOpKind, BinaryOpKind, CompareOpKind};
+    let invalid =
+        || Error::Invalid("unsupported binary operation or invalid checked-division domain".into());
+    if operator.vector_match.is_some() {
+        return Err(invalid());
+    }
+    if operator.checked_relative_division || operator.checked_finite_division {
+        if operator.kind != BinaryOpKind::Arithmetic(ArithmeticOpKind::Div)
+            || !left.is_finite()
+            || !right.is_finite()
+            || right == 0.
+        {
+            return Err(invalid());
+        }
+        let value = left / right;
+        if !value.is_finite() || (operator.checked_relative_division && !value.is_normal()) {
+            return Err(invalid());
+        }
+        return Ok(Value::Float64(value));
+    }
+    Ok(match operator.kind {
+        BinaryOpKind::Arithmetic(ref op) => {
+            Value::Float64(evaluate_float64_arithmetic(op, left, right))
+        }
+        BinaryOpKind::Compare(ref op) => Value::Bool(match op {
+            CompareOpKind::Eq => left == right,
+            CompareOpKind::Ne => left != right,
+            CompareOpKind::Lt => left < right,
+            CompareOpKind::Le => left <= right,
+            CompareOpKind::Gt => left > right,
+            CompareOpKind::Ge => left >= right,
+            _ => return Err(invalid()),
+        }),
+        _ => return Err(invalid()),
+    })
+}