diff --git a/crates/asap-aware-mapping/src/query_physical_lowering.rs b/crates/asap-aware-mapping/src/query_physical_lowering.rs
index 39d04ced..bd2aa675 100644
--- a/crates/asap-aware-mapping/src/query_physical_lowering.rs
+++ b/crates/asap-aware-mapping/src/query_physical_lowering.rs
@@ -1187,6 +1187,7 @@ fn supports_hash_aggregate(
                     | AggIntent::Avg { .. }
                     | AggIntent::StdDev { .. }
                     | AggIntent::Variance { .. }
+                    | AggIntent::PearsonCorr { .. }
                     | AggIntent::Group
                     | AggIntent::CountValues { .. }
             )
@@ -1443,6 +1444,56 @@ mod tests {
         ));
     }
 
+    // Correlation can be costed as an exact hash aggregate using provider-supplied state size.
+    #[test]
+    fn correlation_lowers_to_physical_hash_aggregate() {
+        use asap_types::pre_asap::{
+            AggIntent, Column, DataType, QueryExpr, Reduction, Schema, Source,
+        };
+        let source = Source::Table {
+            table_ref: "pairs".into(),
+        };
+        let root = Rc::new(QueryExpr::Aggregate {
+            reduction: Reduction::by(vec![]),
+            measures: vec![AggIntent::PearsonCorr { left: 0, right: 1 }],
+            output_names: vec!["r".into()],
+            having: None,
+            child: Rc::new(QueryExpr::Scan {
+                source: source.clone(),
+                predicates: vec![],
+                schema: Schema::new(vec![
+                    Column::new("x", DataType::Float64, true),
+                    Column::new("y", DataType::Float64, true),
+                ]),
+            }),
+        });
+        let scope = scope(vec![coverage(source, vec![])]);
+        let provided = HashMap::from([
+            (
+                "query-1".into(),
+                evidence(scan_stats(edge(100, 1600), 1600)),
+            ),
+            (
+                "query-0".into(),
+                evidence(OperatorStatistics::HashAggregate {
+                    edges: unary_edges(edge(100, 1600), edge(1, 8)),
+                    group_count: 1,
+                    key_bytes: 0,
+                    accumulator_bytes_per_group: 48,
+                }),
+            ),
+        ]);
+        let dag = lower_query_physical_dag(&root, &scope, &scripted(&provided)).unwrap();
+        assert!(matches!(
+            dag.nodes.last().unwrap().operator,
+            PhysicalOperator::HashAggregate {
+                grouping_key_count: 0,
+                accumulator_count: 1,
+            }
+        ));
+        assert!(estimate_physical_dag(&dag.nodes, &dag.root, &scope, &dag.evidence).is_ok());
+    }
+
     #[test]
     fn query_lowering_recurses_and_fuses_global_sort_limit() {
         use asap_types::pre_asap::{AggIntent, GroupKeys, QueryExpr, Reduction, SortKey, Source};
diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs
index 075f13a1..13cffa35 100644
--- a/crates/asap-aware-mapping/src/replacement.rs
+++ b/crates/asap-aware-mapping/src/replacement.rs
@@ -819,7 +819,10 @@ pub(crate) fn implementations_for_with(
 
         // ── Exact, non-mergeable reducers — richer partial state than a
         //    single value (see `agg_is_mergeable`), so no accumulator form.
-        AggIntent::Avg { .. } | AggIntent::StdDev { .. } | AggIntent::Variance { .. } => {
+        AggIntent::Avg { .. }
+        | AggIntent::StdDev { .. }
+        | AggIntent::Variance { .. }
+        | AggIntent::PearsonCorr { .. } => {
             vec![Implementation::PassThrough]
         }
 
@@ -6062,6 +6065,17 @@ mod tests {
         }
     }
 
+    // Correlation must never acquire a single-input sketch or scalar accumulator.
+    #[test]
+    fn pearson_corr_keeps_exact_paired_input() {
+        let intent = AggIntent::PearsonCorr { left: 0, right: 1 };
+        assert!(matches!(
+            implementations_for_with(&intent, &crate::cost_model::DefaultCostModel).as_slice(),
+            [Implementation::PassThrough]
+        ));
+        assert!(summary_candidates(&intent).is_empty());
+    }
+
     #[test]
     fn accuracy_target_drives_the_boundary() {
         // Same intent, three targets → three different decisions.
diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs
index 5e085a06..0142dd03 100644
--- a/crates/frontend-sql/src/sql/mod.rs
+++ b/crates/frontend-sql/src/sql/mod.rs
@@ -1653,6 +1653,25 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
                 reducer_col(&name, args).map(Some)
             };
             Ok(match semantic {
+                AggSemantic::Correlation => {
+                    if agg_fn.filter.is_some()
+                        || agg_fn.order_by.is_some()
+                        || agg_fn.null_treatment.is_some()
+                    {
+                        return Err(LoweringError::UnsupportedAggregate(
+                            "corr with FILTER, ORDER BY, or explicit null treatment".into(),
+                        ));
+                    }
+                    let [left, right] = agg_fn.args.as_slice() else {
+                        return Err(LoweringError::UnsupportedAggregate(
+                            "corr requires two arguments".into(),
+                        ));
+                    };
+                    AggIntent::PearsonCorr {
+                        left: expr_to_group_ref(left)?,
+                        right: expr_to_group_ref(right)?,
+                    }
+                }
                 AggSemantic::Count if agg_fn.distinct => AggIntent::Cardinality {
                     col: col(&agg_fn.args)?,
                     accuracy: current_accuracy(),
@@ -2051,6 +2070,21 @@ impl DerivedCols {
         let Expr::AggregateFunction(agg_fn) = unalias(expr) else {
             return Ok(expr.clone());
         };
+        if matches!(
+            asap_sql_function_catalog::lookup_native(&agg_fn.func.name().to_lowercase()),
+            Some(AggSemantic::Correlation)
+        ) {
+            // Give each value argument its own projected name, including casts
+            // and qualified columns. This retains both inputs and avoids losing
+            // relation qualifiers when the projection becomes an unqualified schema.
+            let mut rewritten = agg_fn.clone();
+            for arg in &mut rewritten.args {
+                let alias = unalias(arg).to_string();
+                self.materialize(alias.clone(), df_expr_to_unresolved(arg)?)?;
+                *arg = Expr::Column(DfColumn::new_unqualified(alias));
+            }
+            return Ok(Expr::AggregateFunction(rewritten));
+        }
         // `COUNT(*)` reduces no column; `agg_col_name` covers bare/aliased/cast
         // columns, so `None` here means the argument really is an expression.
         let counts_rows = agg_fn.func.name().eq_ignore_ascii_case("count") && !agg_fn.distinct;
diff --git a/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs
index e6b0c83c..ab370ce3 100644
--- a/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs
+++ b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs
@@ -91,13 +91,11 @@ async fn lowers_the_warehouse_ingestion_check_set() {
             Err(error) => panic!("unexpected failure for U-{id}: {error}"),
         }
     }
+    // P4d and P4l are Pearson correlation checks; they lower since `corr`
+    // became `AggIntent::PearsonCorr`.
     assert_eq!(
         rejected,
-        vec![
-            ("P2b", "multi-column COUNT(DISTINCT)".into()),
-            ("P4d", "corr".into()),
-            ("P4l", "corr".into()),
-        ]
+        vec![("P2b", "multi-column COUNT(DISTINCT)".into())]
     );
-    assert_eq!(lowered, 47);
+    assert_eq!(lowered, 49);
 }
diff --git a/crates/frontend-sql/tests/pearson_corr.rs b/crates/frontend-sql/tests/pearson_corr.rs
new file mode 100644
index 00000000..82e9d60d
--- /dev/null
+++ b/crates/frontend-sql/tests/pearson_corr.rs
@@ -0,0 +1,136 @@
+//! Correlation retains both arguments through SQL lowering and exact planning.
+use std::rc::Rc;
+
+use asap_frontend_sql::{lower_sql, SqlCatalog};
+use asap_types::pre_asap::{AggIntent, Column, DataType, QueryExpr, Schema};
+use asap_types::types::AccuracyTarget;
+
+fn catalog() -> SqlCatalog {
+    let schema = Schema::new(vec![
+        Column::new("x", DataType::Float64, true),
+        Column::new("y", DataType::Float64, true),
+        Column::new("g", DataType::Int64, false),
+    ]);
+    SqlCatalog::new()
+        .with_table("a", schema.clone())
+        .with_table("b", schema)
+}
+
+async fn lower(sql: &str) -> QueryExpr {
+    lower_sql(sql, &catalog(), AccuracyTarget::Exact)
+        .await
+        .unwrap()
+}
+
+fn aggregate(query: &QueryExpr) -> (&[AggIntent], &QueryExpr) {
+    match query {
+        QueryExpr::Aggregate {
+            measures, child, ..
+        } => (measures, child),
+        QueryExpr::Project { child, .. }
+        | QueryExpr::Filter { child, .. }
+        | QueryExpr::Sort { child, .. } => aggregate(child),
+        other => panic!("expected aggregate, got {other:?}"),
+    }
+}
+
+// Expressions in either argument, including casts, survive as projected values.
+#[tokio::test]
+async fn corr_materializes_both_arguments() {
+    for sql in [
+        "SELECT corr(x + 1, y * 2) FROM a",
+        "SELECT corr(x, y * 2) FROM a",
+        "SELECT corr(CAST(g AS DOUBLE), y) FROM a",
+        "SELECT corr(x, 1.0) FROM a",
+    ] {
+        let query = lower(sql).await;
+        let (measures, child) = aggregate(&query);
+        assert_eq!(measures, &[AggIntent::PearsonCorr { left: 0, right: 1 }]);
+        let QueryExpr::Project { cols, .. } = child else {
+            panic!("derived inputs")
+        };
+        assert_eq!(cols.len(), 2);
+        assert!(cols
+            .iter()
+            .any(|col| !matches!(col.expr, QueryExpr::Column(_))));
+        assert_eq!(
+            query.output_schema().unwrap().columns[0].dtype,
+            DataType::Float64
+        );
+    }
+}
+
+// Identically named join columns remain distinct even when projected beneath the aggregate.
+#[tokio::test]
+async fn corr_preserves_qualified_join_inputs() {
+    let query = lower("SELECT corr(a.x, b.x) FROM a JOIN b ON a.g = b.g").await;
+    let (measures, child) = aggregate(&query);
+    assert_eq!(measures[0].input_cols(), vec![0, 1]);
+    let QueryExpr::Project { cols, .. } = child else {
+        panic!("paired projection")
+    };
+    assert_eq!(cols[0].expr, QueryExpr::Column(0));
+    assert_eq!(cols[1].expr, QueryExpr::Column(3));
+}
+
+// Grouping and sibling reducers cannot drop either correlation argument.
+#[tokio::test]
+async fn corr_coexists_with_grouping_having_and_other_measures() {
+    let query = lower("SELECT g, corr(x, y) AS r, sum(x * y) AS s FROM a GROUP BY g HAVING corr(x, y) > 0 ORDER BY r").await;
+    let (measures, child) = aggregate(&query);
+    let pair = measures
+        .iter()
+        .find(|m| matches!(m, AggIntent::PearsonCorr { .. }))
+        .unwrap();
+    let schema = child.output_schema().unwrap();
+    for id in pair.input_cols() {
+        assert!(id < schema.columns.len());
+    }
+    assert!(measures.iter().any(|m| matches!(m, AggIntent::Sum { .. })));
+    let output = query.output_schema().unwrap();
+    assert_eq!(output.columns[1].name, "r");
+    assert_eq!(output.columns[1].dtype, DataType::Float64);
+    assert!(output.columns[1].nullable);
+}
+
+// Repeated inputs reuse their value while retaining two argument positions.
+#[tokio::test]
+async fn corr_repeated_input_and_serialization() {
+    let query = lower("SELECT corr(x, x) FROM a").await;
+    assert_eq!(aggregate(&query).0[0].input_cols(), vec![0, 0]);
+    let encoded = serde_json::to_string(&query).unwrap();
+    let decoded: QueryExpr = serde_json::from_str(&encoded).unwrap();
+    assert_eq!(query, decoded);
+}
+
+// Unsupported modifiers and window calls fail instead of silently changing semantics.
+#[tokio::test]
+async fn corr_rejects_unrepresented_forms() {
+    for sql in [
+        "SELECT corr(DISTINCT x, y) FROM a",
+        "SELECT corr(x, y) FILTER (WHERE g > 0) FROM a",
+        "SELECT corr(x, y ORDER BY g) FROM a",
+        "SELECT corr(x, y) OVER () FROM a",
+        "SELECT corr(x) FROM a",
+    ] {
+        assert!(
+            lower_sql(sql, &catalog(), AccuracyTarget::Exact)
+                .await
+                .is_err(),
+            "{sql}"
+        );
+    }
+}
+
+// Exact fallback retains the complete typed query and compiles to an executable DAG.
+#[tokio::test]
+async fn corr_survives_exact_plan_compilation() {
+    let query = Rc::new(lower("SELECT corr(x, y) AS r FROM a").await);
+    let plan = asap_aware_mapping::replacement::keep_pre_asap(&query).unwrap();
+    assert!(plan.guarantee.as_ref().unwrap().is_exact());
+    let asap_types::post_asap::SummaryExpr::KeepPreAsap(retained) = &plan.expr else {
+        panic!("expected exact fallback");
+    };
+    assert_eq!(aggregate(retained).0, aggregate(&query).0);
+    asap_types::post_asap::compile_executable_dag(&plan).unwrap();
+}
diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs
index 642c9713..0dc17b2c 100644
--- a/crates/frontend-sql/tests/sql_lowering.rs
+++ b/crates/frontend-sql/tests/sql_lowering.rs
@@ -2455,6 +2455,16 @@ async fn clickhouse_tuple_element_preserves_declared_field_metadata() {
     }
 }
 
+/// Correlation lowers to a nullable numeric result instead of UnsupportedAggregate.
+#[tokio::test]
+async fn corr_result_is_nullable_float() {
+    let query = lower("SELECT corr(latency, bytes) AS correlation FROM metrics").await;
+    let schema = query.output_schema().unwrap();
+    assert_eq!(schema.columns[0].name, "correlation");
+    assert_eq!(schema.columns[0].dtype, DataType::Float64);
+    assert!(schema.columns[0].nullable);
+}
+
 // A multi-column DISTINCT must not silently count only the first column.
 #[tokio::test]
 async fn composite_distinct_is_rejected() {
diff --git a/crates/sql-function-catalog/src/lib.rs b/crates/sql-function-catalog/src/lib.rs
index 8991a8a6..df2c0926 100644
--- a/crates/sql-function-catalog/src/lib.rs
+++ b/crates/sql-function-catalog/src/lib.rs
@@ -91,6 +91,8 @@ pub enum AggSemantic {
     Min,
     Max,
     Avg,
+    /// Pearson correlation of two numeric value arguments.
+    Correlation,
     /// Sample stddev unless `population`.
     StdDev {
         population: bool,
@@ -126,6 +128,11 @@ pub struct NativeFunction {
 /// when DataFusion (or this front end) accepts more than one spelling for
 /// the same semantic (`avg`/`mean`, `stddev`/`stddev_samp`, ...).
 pub const NATIVE_FUNCTIONS: &[NativeFunction] = &[
+    NativeFunction {
+        name: "corr",
+        arity: Arity::Exact(2),
+        semantic: AggSemantic::Correlation,
+    },
     NativeFunction {
         name: "count",
         arity: Arity::Range { min: 0, max: 1 },
@@ -311,7 +318,7 @@ pub const CLICKHOUSE_BUILTINS: &[ClickHouseBuiltin] = &[
 /// Aggregate function names DataFusion's own planner resolves out of the box
 /// that this catalog deliberately does *not* map to a canonical
 /// [`AggSemantic`] -- either because `AggIntent` has no shape for them
-/// (a multi-column correlation/regression aggregate, a bitwise/boolean
+/// (an unimplemented covariance/regression aggregate, a bitwise/boolean
 /// aggregate, a string concatenation aggregate, ...) or because this front
 /// end already rejects them explicitly elsewhere (`array_agg`, `grouping`).
 ///
@@ -345,10 +352,8 @@ pub const KNOWN_UNMAPPED_NATIVE_FUNCTIONS: &[&str] = &[
     "bit_xor",
     "bool_and",
     "bool_or",
-    // Two-column correlation / linear-regression aggregates -- every
-    // `AggIntent` value reducer takes one input column (`reducer_col` in
-    // `asap-frontend-sql`), so these have no home yet.
-    "corr",
+    // Two-column covariance / regression operations with no `AggIntent`
+    // variant; `AggIntent::PearsonCorr` covers `corr` only.
     "covar",
     "covar_pop",
     "covar_samp",
diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs
index 2dcd15b3..26b30e63 100644
--- a/crates/types/src/post_asap/execution_data_state.rs
+++ b/crates/types/src/post_asap/execution_data_state.rs
@@ -647,13 +647,11 @@ fn check_plain_operands(
         .map(|keys| keys.keys().to_vec())
         .unwrap_or_default();
     for m in measures {
-        if let Some(col) = m.input_col() {
-            referenced.push(col);
-        }
+        referenced.extend(m.input_cols());
     }
     // With no explicit input column (the PromQL sample-value convention)
     // the operator reads every non-key column, so all must be plain.
-    let implicit = measures.iter().any(|m| m.input_col().is_none());
+    let implicit = measures.iter().any(|m| m.input_cols().is_empty());
     for (i, field) in input.fields.iter().enumerate() {
         if !(implicit || referenced.contains(&i)) {
             continue;
@@ -1069,6 +1067,28 @@ mod tests {
         assert!(validate_execution_data_states(&root).is_err());
     }
 
+    // Both paired operands must be plain; an unrelated state column is not an input.
+    #[test]
+    fn pearson_corr_checks_both_operand_states() {
+        let operation = ValueOperation::Exact(ExactOperation::Aggregate {
+            reduction: Reduction::by(vec![]),
+            measures: vec![AggIntent::PearsonCorr { left: 0, right: 1 }],
+            output_names: vec![],
+            having: None,
+        });
+        for operand in [0, 1] {
+            let mut input = plain(&["x", "y", "unused"]);
+            input.fields[operand].dtype = kll();
+            assert!(matches!(
+                check_plain_operands(&operation, &input),
+                Err(ExecutionDataStateError::NonPlainOperand { .. })
+            ));
+        }
+        let mut input = plain(&["x", "y", "unused"]);
+        input.fields[2].dtype = kll();
+        check_plain_operands(&operation, &input).unwrap();
+    }
+
     #[test]
     fn exact_operator_schema_matches_pre_asap_aggregate_derivation() {
         let child_schema = lift_plain(&scan().output_schema().unwrap());
diff --git a/crates/types/src/pre_asap/agg_intent.rs b/crates/types/src/pre_asap/agg_intent.rs
index a3ae5ceb..6b020c74 100644
--- a/crates/types/src/pre_asap/agg_intent.rs
+++ b/crates/types/src/pre_asap/agg_intent.rs
@@ -73,6 +73,13 @@ pub enum AggIntent<C = ColumnId> {
         col: Option<C>,
         population: bool,
     },
+    /// Pearson correlation — SQL `CORR(left, right)`. The only aggregate with
+    /// two value inputs; both references take part in binding and dependency
+    /// tracking, so neither is reachable through `input_col`.
+    PearsonCorr {
+        left: C,
+        right: C,
+    },
     /// φ-quantile of `col`. SQL `approx_percentile_cont(col, φ)` and
     /// `median(col)` (φ=0.5); PromQL `quantile(φ, …)` leaves `col` as `None`
     /// (the sample value).
@@ -468,11 +475,19 @@ impl<C: Clone> AggIntent<C> {
 }
 
 impl<C: Clone> AggIntent<C> {
-    /// The input column this intent reduces, if it carries one. `None` = the
-    /// synthetic time-series sample value (PromQL) or an argument-less
-    /// aggregate (`Count` / `TopK`). Used by schema derivation to resolve each
-    /// reducer's input column, and by `plan::bind` to pick the column a
-    /// summary is built over.
+    /// All explicit value-column dependencies, in argument order.
+    /// An empty list retains the existing implicit sample/row-count convention.
+    pub fn input_cols(&self) -> Vec<C> {
+        match self {
+            Self::PearsonCorr { left, right } => vec![left.clone(), right.clone()],
+            _ => self.input_col().into_iter().collect(),
+        }
+    }
+
+    /// The explicit input of a single-column reducer. `PearsonCorr`,
+    /// argument-less aggregates, and implicit PromQL sample inputs return `None`.
+    /// Use `input_cols` for dependency tracking; this accessor is for consumers
+    /// that have already selected a single-column implementation.
     pub fn input_col(&self) -> Option<C> {
         match self {
             AggIntent::Sum { col }
@@ -504,6 +519,9 @@ impl<C: Clone> AggIntent<C> {
             AggIntent::Avg { .. } => col("avg", DataType::Float64, false),
             AggIntent::StdDev { .. } => col("stddev", DataType::Float64, false),
             AggIntent::Variance { .. } => col("variance", DataType::Float64, false),
+            // Nullable: correlation is undefined for fewer than two pairs where
+            // both inputs are non-null, and for a zero-variance input.
+            AggIntent::PearsonCorr { .. } => col("corr", DataType::Float64, true),
             AggIntent::Quantile { q, .. } => col(
                 &format!("quantile_{}", quantile_suffix(*q)),
                 DataType::Float64,
@@ -628,16 +646,20 @@ pub mod topk {
 
 /// Two instances of this aggregation can be merged
 /// (`agg(A ∪ B) = combine(agg(A), agg(B))`). `Avg` / `StdDev` / `Variance`
-/// need richer partial state than a single value, so they are not mergeable.
+/// and correlation need richer partial state than a single value, so
+/// their finalized values are not mergeable.
 pub fn agg_is_mergeable(op: &AggIntent) -> bool {
     !matches!(
         op,
-        AggIntent::Avg { .. } | AggIntent::StdDev { .. } | AggIntent::Variance { .. }
+        AggIntent::Avg { .. }
+            | AggIntent::StdDev { .. }
+            | AggIntent::Variance { .. }
+            | AggIntent::PearsonCorr { .. }
     )
 }
 
 /// Whether this op implies `exact_required` — no sketch benefit. The exact
-/// intents are `Sum / Count / Avg / Min / Max`.
+/// intents include `Sum / Count / Avg / Min / Max` and correlation.
 pub fn agg_is_exact(op: &AggIntent) -> bool {
     matches!(
         op,
@@ -646,6 +668,7 @@ pub fn agg_is_exact(op: &AggIntent) -> bool {
             | AggIntent::Avg { .. }
             | AggIntent::Min { .. }
             | AggIntent::Max { .. }
+            | AggIntent::PearsonCorr { .. }
             | AggIntent::Group
             | AggIntent::CountValues { .. }
     )
@@ -700,6 +723,22 @@ mod tests {
         Column::new(name, dtype, false)
     }
 
+    // Correlation exposes both dependencies but cannot merge final scalar results.
+    #[test]
+    fn pearson_corr_contract() {
+        let intent = AggIntent::PearsonCorr { left: 2, right: 5 };
+        assert_eq!(intent.input_cols(), vec![2, 5]);
+        assert_eq!(intent.input_col(), None);
+        assert!(agg_is_exact(&intent));
+        assert!(!agg_is_mergeable(&intent));
+        let output = intent.output_column(&c("x", DataType::Float64));
+        assert_eq!(output.dtype, DataType::Float64);
+        assert!(output.nullable);
+        let value = serde_json::to_value(&intent).unwrap();
+        assert_eq!(value["kind"], "pearson_corr");
+        assert_eq!(serde_json::from_value::<AggIntent>(value).unwrap(), intent);
+    }
+
     #[test]
     fn output_column_names_are_intent_keyed() {
         let v = c("value", DataType::Float64);
diff --git a/crates/types/src/pre_asap/binder.rs b/crates/types/src/pre_asap/binder.rs
index 221aebdb..55483841 100644
--- a/crates/types/src/pre_asap/binder.rs
+++ b/crates/types/src/pre_asap/binder.rs
@@ -203,17 +203,14 @@ pub(crate) fn collect_referenced_columns(tree: &UnresolvedQueryExpr) -> Vec<Stri
             push_ref_name(c, out);
         }
     }
-    fn opt_ref(c: &Option<ColumnRef>, out: &mut Vec<String>) {
-        if let Some(c) = c {
-            push_ref_name(c, out);
-        }
-    }
     fn group_keys(g: &super::query_expr::GroupKeys<ColumnRef>, out: &mut Vec<String>) {
         g.keys().iter().for_each(|k| push_ref_name(k, out));
     }
     fn measure_cols(measures: &[super::agg_intent::AggIntent<ColumnRef>], out: &mut Vec<String>) {
         for m in measures {
-            opt_ref(&m.input_col(), out);
+            for c in m.input_cols() {
+                push_ref_name(&c, out);
+            }
         }
     }
     fn walk(node: &UnresolvedQueryExpr, out: &mut Vec<String>) {
@@ -381,6 +378,26 @@ mod tests {
         }
     }
 
+    // Both correlation inputs must seed a usage-derived schema before positional resolution.
+    #[test]
+    fn pearson_corr_inputs_seed_usage_derived_schema() {
+        use crate::pre_asap::{AggIntent, Reduction};
+        let tree = UnresolvedQueryExpr::Aggregate {
+            reduction: Reduction::by(vec![]),
+            measures: vec![AggIntent::PearsonCorr {
+                left: ColumnRef::Named("x".into()),
+                right: ColumnRef::Named("y".into()),
+            }],
+            output_names: vec![],
+            having: None,
+            child: Rc::new(src("m")),
+        };
+        assert_eq!(collect_referenced_columns(&tree), vec!["x", "y"]);
+        let schema = Binder::new().bind(&tree);
+        assert!(schema.column_id("x").is_some());
+        assert!(schema.column_id("y").is_some());
+    }
+
     #[test]
     fn bare_source_yields_ts_value_floor() {
         let schema = Binder::new().bind(&src("m"));
diff --git a/crates/types/src/pre_asap/resolve.rs b/crates/types/src/pre_asap/resolve.rs
index bfd0f1bf..5d1e722c 100644
--- a/crates/types/src/pre_asap/resolve.rs
+++ b/crates/types/src/pre_asap/resolve.rs
@@ -518,6 +518,10 @@ fn resolve_agg_intent(
         AggIntent::Count { accuracy } => AggIntent::Count {
             accuracy: accuracy.clone(),
         },
+        AggIntent::PearsonCorr { left, right } => AggIntent::PearsonCorr {
+            left: resolve_column_ref(left, schema)?,
+            right: resolve_column_ref(right, schema)?,
+        },
         AggIntent::Sum { col: c } => AggIntent::Sum { col: col(c)? },
         AggIntent::Min { col: c } => AggIntent::Min { col: col(c)? },
         AggIntent::Max { col: c } => AggIntent::Max { col: col(c)? },
@@ -609,6 +613,38 @@ mod tests {
         BinaryOpKind, QueryExpr, Source, VectorMatch, VectorMatchKind,
     };
 
+    // Both sides resolve with qualifiers; an unknown right input is an error.
+    #[test]
+    fn resolve_pearson_corr_inputs() {
+        use crate::pre_asap::{Column, DataType};
+        let schema = Schema::new(vec![
+            Column::new("x", DataType::Float64, true).with_table("a"),
+            Column::new("x", DataType::Float64, true).with_table("b"),
+        ]);
+        let intent = AggIntent::PearsonCorr {
+            left: ColumnRef::Qualified {
+                table: "a".into(),
+                name: "x".into(),
+            },
+            right: ColumnRef::Qualified {
+                table: "b".into(),
+                name: "x".into(),
+            },
+        };
+        assert_eq!(
+            resolve_agg_intent(&intent, &schema).unwrap(),
+            AggIntent::PearsonCorr { left: 0, right: 1 }
+        );
+        let missing = AggIntent::PearsonCorr {
+            left: ColumnRef::Qualified {
+                table: "a".into(),
+                name: "x".into(),
+            },
+            right: ColumnRef::Named("missing".into()),
+        };
+        assert!(resolve_agg_intent(&missing, &schema).is_err());
+    }
+
     /// `resolve_root` over a `BinaryOp { <vector>, PromqlScalarBridge, vector_match }`
     /// (issue #220): the bridged scalar operand resolves through the same
     /// generic walk as every other node (its `Literal` child has no
diff --git a/docs/design_docs/pre-asap-ir.md b/docs/design_docs/pre-asap-ir.md
index 67b325c0..4bb52922 100644
--- a/docs/design_docs/pre-asap-ir.md
+++ b/docs/design_docs/pre-asap-ir.md
@@ -104,7 +104,7 @@ native-histogram accessors — this list is representative, not exhaustive:
 
 ```text
 Count, Sum(col), Min(col), Max(col), Avg(col), StdDev(col), Variance(col),
-Quantile(col, q), TopK(k), Cardinality(col)                      // data-model-agnostic
+Quantile(col, q), TopK(k), Cardinality(col), PearsonCorr(left, right)  // data-model-agnostic
 Rate, Increase                                                    // counter derivatives
 Changes, Delta, IDelta, Deriv, Resets,
 PredictLinear(seconds), DoubleExpSmoothing(sf, tf)                // range-vector functions
@@ -113,6 +113,22 @@ HistogramStdVar, HistogramFraction(lo, hi), HistogramQuantile(q)  // native-hist
 Math(func)                                                        // element-wise transform
 ```
 
+`PearsonCorr { left, right }` is the only measure with two value inputs. Both
+references resolve to positional column IDs, and `input_cols()` exposes both
+dependencies — `input_col()` returns `None`, so single-column consumers cannot
+pick up half of the pair. SQL lowering projects both arguments, preserving
+expressions, casts, and qualified join columns. The result is nullable `Float64`,
+with pairwise null handling owned by the executing engine. It remains exact:
+finalized correlation coefficients cannot be combined as scalar rollups, and no
+sketch or maintained correlation accumulator is selected. Physical costing accepts
+it as a hash aggregate with provider-supplied accumulator size.
+
+SQL `corr` currently rejects `DISTINCT`, aggregate `FILTER`, aggregate `ORDER BY`,
+explicit null treatment, and window usage (`OVER`). Further two-input statistics
+(`covar`, the `regr_*` family) would each add their own variant, following the
+`StdDev` / `Variance` precedent, once they have explicit output and realization
+semantics.
+
 Additional measures can be added when there is a stable semantic distinction and a
 meaningful summary implementation.