feat(SQL): L1 to L3 SQL lowering via DataFusion with TDD

Cargo.toml

@@ -1,5 +1,6 @@
 [workspace]
 members = [
     "crates/core",
+    "crates/lower",
 ]
 resolver = "2"

crates/core/src/intent_algebra/expr_ir.rs

@@ -0,0 +1,173 @@
+use super::expr::ColumnRef;
+use super::schema::L3DataType;
+
+// ── Scalar literals ───────────────────────────────────────────────────────────
+
+/// A typed scalar constant. Used in `L3Expr::Literal`.
+#[derive(Debug, Clone, PartialEq)]
+pub enum L3Scalar {
+    Int64(i64),
+    Float64(f64),
+    Utf8(String),
+    Boolean(bool),
+    Null,
+}
+
+// ── Comparison operators ──────────────────────────────────────────────────────
+
+/// Binary comparison operators for `L3Expr::Compare`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CompareOp {
+    Eq,
+    Ne,
+    Lt,
+    Le,
+    Gt,
+    Ge,
+    Like,
+    NotLike,
+    ILike,
+    NotILike,
+}
+
+// ── Arithmetic operators ──────────────────────────────────────────────────────
+
+/// Binary arithmetic operators for `L3Expr::Arith`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ArithOp {
+    Add,
+    Sub,
+    Mul,
+    Div,
+    Mod,
+}
+
+// ── Expression IR ─────────────────────────────────────────────────────────────
+
+/// A scalar expression used in filter predicates, projection lists, and sort
+/// keys. Flat conjunctions (`BoolAnd`) and disjunctions (`BoolOr`) make
+/// per-conjunct selectivity estimation and PromQL label-matcher lowering
+/// straightforward without recursive descent.
+#[derive(Debug, Clone, PartialEq)]
+pub enum L3Expr {
+    /// Reference to a named column.
+    Column(ColumnRef),
+    /// A constant literal value.
+    Literal(L3Scalar),
+    /// `left op right` — binary comparison.
+    Compare {
+        left: Box<L3Expr>,
+        op: CompareOp,
+        right: Box<L3Expr>,
+    },
+    /// Flat conjunction (logical AND). An empty list is vacuously true.
+    BoolAnd(Vec<L3Expr>),
+    /// Flat disjunction (logical OR). An empty list is vacuously false.
+    BoolOr(Vec<L3Expr>),
+    /// Logical NOT.
+    Not(Box<L3Expr>),
+    /// `expr IS NULL`.
+    IsNull(Box<L3Expr>),
+    /// `expr IS NOT NULL`.
+    IsNotNull(Box<L3Expr>),
+    /// `CAST(expr AS to)`. `try_cast` is `true` for SQL `TRY_CAST`, which
+    /// returns `NULL` on conversion failure instead of raising an error.
+    Cast {
+        expr: Box<L3Expr>,
+        to: L3DataType,
+        try_cast: bool,
+    },
+    /// `expr [NOT] IN (v1, v2, …)`.
+    InList {
+        expr: Box<L3Expr>,
+        list: Vec<L3Expr>,
+        negated: bool,
+    },
+    /// Scalar function call, e.g. `LOWER(col)`, `ABS(x)`.
+    FunctionCall { name: String, args: Vec<L3Expr> },
+    /// Binary arithmetic: `left op right`.
+    Arith {
+        op: ArithOp,
+        left: Box<L3Expr>,
+        right: Box<L3Expr>,
+    },
+    /// SQL `CASE` expression (both searched and simple forms).
+    /// `operand` is present for simple CASE (`CASE expr WHEN ...`),
+    /// absent for searched CASE (`CASE WHEN condition THEN ...`).
+    Case {
+        operand: Option<Box<L3Expr>>,
+        branches: Vec<(L3Expr, L3Expr)>,
+        else_expr: Option<Box<L3Expr>>,
+    },
+}
+
+impl L3Expr {
+    /// If this expression is a `BoolAnd`, return its elements.
+    /// Otherwise return a single-element slice containing `self`.
+    /// Lets callers iterate all top-level conjuncts without cloning.
+    pub fn conjuncts(&self) -> &[L3Expr] {
+        match self {
+            L3Expr::BoolAnd(v) => v.as_slice(),
+            _ => std::slice::from_ref(self),
+        }
+    }
+
+    /// If this expression is a `BoolOr`, return its elements.
+    /// Otherwise return a single-element slice containing `self`.
+    pub fn disjuncts(&self) -> &[L3Expr] {
+        match self {
+            L3Expr::BoolOr(v) => v.as_slice(),
+            _ => std::slice::from_ref(self),
+        }
+    }
+
+    /// Recursively collect all `ColumnRef`s referenced anywhere in this
+    /// expression. Used by L4 for column-lineage and selectivity estimation.
+    pub fn columns_referenced(&self) -> Vec<&ColumnRef> {
+        match self {
+            L3Expr::Column(c) => vec![c],
+            L3Expr::Literal(_) => vec![],
+            L3Expr::Compare { left, right, .. } => {
+                let mut v = left.columns_referenced();
+                v.extend(right.columns_referenced());
+                v
+            }
+            L3Expr::BoolAnd(parts) | L3Expr::BoolOr(parts) => {
+                parts.iter().flat_map(|e| e.columns_referenced()).collect()
+            }
+            L3Expr::Not(e) | L3Expr::IsNull(e) | L3Expr::IsNotNull(e) => e.columns_referenced(),
+            L3Expr::Cast { expr, .. } => expr.columns_referenced(),
+            L3Expr::InList { expr, list, .. } => {
+                let mut v = expr.columns_referenced();
+                v.extend(list.iter().flat_map(|e| e.columns_referenced()));
+                v
+            }
+            L3Expr::FunctionCall { args, .. } => {
+                args.iter().flat_map(|e| e.columns_referenced()).collect()
+            }
+            L3Expr::Arith { left, right, .. } => {
+                let mut v = left.columns_referenced();
+                v.extend(right.columns_referenced());
+                v
+            }
+            L3Expr::Case {
+                operand,
+                branches,
+                else_expr,
+            } => {
+                let mut v = vec![];
+                if let Some(op) = operand {
+                    v.extend(op.columns_referenced());
+                }
+                for (when, then) in branches {
+                    v.extend(when.columns_referenced());
+                    v.extend(then.columns_referenced());
+                }
+                if let Some(e) = else_expr {
+                    v.extend(e.columns_referenced());
+                }
+                v
+            }
+        }
+    }
+}

crates/core/src/intent_algebra/mod.rs

@@ -1,9 +1,11 @@
 pub mod expr;
+pub mod expr_ir;
 pub mod schema;
 
 pub use expr::{
     AggIntent, BinaryOpKind, ColumnRef, DataModel, GroupKey, JoinKey, JoinKind, L3Node,
     LabelFilter, MetricRef, PartitionKeys, Predicate, ProjectItem, QueryExpr, SetOpKind, SortKey,
     Source, TableRef, TimeRange, TimeWindowKind, VectorMatch, WindowFrame, WindowFuncKind,
 };
-pub use schema::{HasSchema, L3DataType, L3Field, L3Schema, SchemaCatalog};
+pub use expr_ir::{ArithOp, CompareOp, L3Expr, L3Scalar};
+pub use schema::{ColumnDef, HasSchema, L3DataType, L3Field, L3Schema, SchemaCatalog, TableSchema};

crates/core/src/intent_algebra/schema.rs

@@ -1,8 +1,45 @@
-/// Opaque handle to the external data-source catalog (Prometheus metric
-/// metadata, SQL `information_schema`, DataFusion catalog). Used only by
-/// `Scan` schema derivation to resolve leaf column types; all other nodes
-/// derive their output schemas purely from their input schemas.
-pub struct SchemaCatalog;
+use std::collections::HashMap;
+
+/// Catalog of known relational tables and their column definitions.
+/// Passed to `lower_batch` so the SQL lowerer can resolve table and column
+/// types and identify each table's designated time column.
+#[derive(Debug, Clone, Default)]
+pub struct SchemaCatalog {
+    pub tables: HashMap<String, TableSchema>,
+}
+
+/// Schema for a single relational table.
+#[derive(Debug, Clone)]
+pub struct TableSchema {
+    pub columns: Vec<ColumnDef>,
+    /// Name of the column that holds the row timestamp. When set, WHERE
+    /// predicates on this column are extracted into `Source::Table.time_range`
+    /// rather than left as opaque `Filter` predicates.
+    pub time_column: Option<String>,
+}
+
+impl TableSchema {
+    /// Returns `Err` if `time_column` names a column that does not exist in `columns`.
+    pub fn validate(&self) -> Result<(), String> {
+        if let Some(tc) = &self.time_column {
+            if !self.columns.iter().any(|c| &c.name == tc) {
+                return Err(format!(
+                    "time_column '{tc}' not found in table columns {:?}",
+                    self.columns.iter().map(|c| &c.name).collect::<Vec<_>>()
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// One column in a `TableSchema`.
+#[derive(Debug, Clone)]
+pub struct ColumnDef {
+    pub name: String,
+    pub data_type: L3DataType,
+    pub nullable: bool,
+}
 
 // ── Data types ────────────────────────────────────────────────────────────────
 
@@ -24,7 +61,7 @@ pub enum L3DataType {
 
 // ── Schema ────────────────────────────────────────────────────────────────────
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct L3Field {
     pub name: String,
     pub dtype: L3DataType,
@@ -35,7 +72,7 @@ pub struct L3Field {
 /// flowing between two operators. Type-checked at plan construction time:
 /// a node whose predicate references a column absent from its child's
 /// `L3Schema` is a plan-time error.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct L3Schema {
     pub fields: Vec<L3Field>,
     /// Index into `fields` for the time axis, if any.
@@ -49,5 +86,10 @@ pub struct L3Schema {
 /// its children's output schemas. The `L3Node` wrapper stores the derived
 /// schema so derivation runs once at construction, not on every traversal.
 pub trait HasSchema {
+    /// # Panics
+    /// Panics if a `Scan` node references a table that is not present in
+    /// `catalog`. Callers must ensure every table referenced by the expression
+    /// tree is registered in the catalog before calling this method.
+    /// `lower_batch` enforces this invariant via upfront catalog validation.
     fn output_schema(&self, input_schemas: &[&L3Schema], catalog: &SchemaCatalog) -> L3Schema;
 }

crates/core/src/sketch_algebra/expr.rs

@@ -1,4 +1,5 @@
 use std::rc::Rc;
+use std::sync::Arc;
 
 use super::schema::L4Schema;
 use super::sketch::{SketchQuery, SummaryKind, SummaryParams};
@@ -31,7 +32,7 @@ pub enum SummaryExpr {
     /// Any L3 node that no L4 rule rewrote (e.g. `Filter`, `Project`, `Sort`).
     /// Output schema is the inner L3 node's schema, lifted to `L4Schema`
     /// with all fields as `L4DataType::Primitive`.
-    Logical(Rc<L3Node>),
+    Logical(Arc<L3Node>),
 
     /// Sketch aggregation. L4 chose `sketch` + `params` from the catalog
     /// for `AggIntent` under `DeploymentConstraints`.