Compute corr, the covariances and regr_* in the steps of PostgreSQL

crates/rudb-bind/src/binder.rs

@@ -6505,6 +6505,9 @@ impl<'a> Binder<'a> {
         {
             bound[0] = self.cast_to(only, &LogicalType::Varchar);
         }
+        if self.semantics.unknown_types() == UnknownTypes::Postgres {
+            self.pg_aggregate_unknowns(ast, name, &args, &mut bound)?;
+        }
         let types: Vec<LogicalType> =
             bound.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
         let mut resolved = resolve(name, &types)?;
@@ -6618,11 +6621,31 @@ impl<'a> Binder<'a> {
     /// exact `numeric` sums and answers a `numeric`, and of a `real` or a `double precision` in the
     /// state of `float8_accum` and answers a `double precision`, where the pin answers a DOUBLE
     /// from a Welford state for both. The calls take the names of PostgreSQL's final functions,
-    /// which are the states of [`rudb_kernels`] that keep them. Any other call is as resolved.
+    /// which are the states of [`rudb_kernels`] that keep them. `corr`, the covariances and the
+    /// `regr_*` family keep the state of `float8_regr_accum` under the names of their final
+    /// functions too, which answers null for a constant column where the pin answers NaN. Any
+    /// other call is as resolved.
     fn spread_as_postgres(&self, resolved: Resolved, types: &[LogicalType]) -> Resolved {
         if self.semantics.aggregate_types() != AggregateTypes::Postgres {
             return resolved;
         }
+        let paired = match resolved.name {
+            "corr" => Some("float8_corr"),
+            "covar_pop" => Some("float8_covar_pop"),
+            "covar_samp" => Some("float8_covar_samp"),
+            "regr_avgx" => Some("float8_regr_avgx"),
+            "regr_avgy" => Some("float8_regr_avgy"),
+            "regr_intercept" => Some("float8_regr_intercept"),
+            "regr_r2" => Some("float8_regr_r2"),
+            "regr_slope" => Some("float8_regr_slope"),
+            "regr_sxx" => Some("float8_regr_sxx"),
+            "regr_sxy" => Some("float8_regr_sxy"),
+            "regr_syy" => Some("float8_regr_syy"),
+            _ => None,
+        };
+        if let Some(name) = paired {
+            return Resolved { name, ..resolved };
+        }
         let ([argument], [_]) = (types, &resolved.arguments[..]) else { return resolved };
         let (argument, returns) = match argument {
             // An exact state reads the integers that fit an `i128` as they are.

crates/rudb-bind/src/pgcalls.rs

@@ -335,6 +335,78 @@ impl Binder<'_> {
         })
     }
 
+    /// The arguments of an aggregate call in a PostgreSQL session that has a string literal among
+    /// them, each literal read as the type of the aggregate that `func_get_detail` finds over
+    /// `unknown` for it. `corr(g, 'NaN')` reads the literal as a `double precision`, and
+    /// `avg('4')` is not unique, since no form of `avg` takes text. A literal whose place has a
+    /// polymorphic type, a name that is not only an aggregate, a type that PostgreSQL does not
+    /// have, and a call that finds nothing keep the path of the pin.
+    pub(crate) fn pg_aggregate_unknowns(
+        &mut self,
+        ast: &Ast,
+        written: &str,
+        arguments: &[ast::ExprRef],
+        bound: &mut [ExprRef],
+    ) -> Result<()> {
+        use rudb_pgtypes::{Resolution, oid};
+        let untyped: Vec<bool> = arguments
+            .iter()
+            .map(|&argument| {
+                matches!(ast.expr(argument), ast::Expr::Literal { kind: LiteralKind::String, .. })
+            })
+            .collect();
+        let procs = rudb_pgtypes::procs(written);
+        if !untyped.contains(&true)
+            || procs.is_empty()
+            || procs.iter().any(|proc| proc.kind != b'a')
+            || arguments.len() != bound.len()
+        {
+            return Ok(());
+        }
+        let mut types = Vec::with_capacity(bound.len());
+        let mut oids = Vec::with_capacity(bound.len());
+        for (&argument, &untyped) in bound.iter().zip(&untyped) {
+            let ty = self.plan().expr_type(argument).clone();
+            let oid = match untyped || ty == LogicalType::Null {
+                true => oid::UNKNOWN,
+                false => match exact_oid(&ty) {
+                    Some(oid) => oid,
+                    None => return Ok(()),
+                },
+            };
+            types.push(ty);
+            oids.push(oid);
+        }
+        let call = rudb_pgtypes::Call { args: &oids, names: &[], variadic: false };
+        let candidate = match rudb_pgtypes::resolve_call(written, call) {
+            Resolution::Found(candidate) => candidate,
+            Resolution::NotFound(_) => return Ok(()),
+            Resolution::Ambiguous => {
+                let call = spelled_call(written, &types, &untyped, &[]);
+                let message = format!("function {call} is not unique");
+                return Err(Error::binder(message.clone())
+                    .state(SqlState::AMBIGUOUS_FUNCTION)
+                    .pg(message)
+                    .detail("Could not choose a best candidate function.")
+                    .hint("You might need to add explicit type casts.")
+                    .with_span(self.current_span));
+            }
+        };
+        for (index, &oid) in candidate.args.iter().enumerate() {
+            if !untyped.get(index).copied().unwrap_or(false) {
+                continue;
+            }
+            // A literal that already has the type stays the expression it is, so that a
+            // `DISTINCT` call still finds it among its sort keys.
+            if let Some(ty) = rudb_pgtypes::logical_type(oid)
+                && ty != types[index]
+            {
+                bound[index] = self.argument_as(ast, arguments[index], bound[index], &ty)?;
+            }
+        }
+        Ok(())
+    }
+
     /// The call `written(arguments)` bound to the function of `pg_proc` that PostgreSQL finds for
     /// it, or `None` when the call takes another path here.
     ///

crates/rudb-kernels/src/general.rs

@@ -30,6 +30,7 @@ use crate::histogram::Binned;
 use crate::lttb::{Plot, Points};
 use crate::number::{approximate, fit, integral};
 use crate::pgorderedset::OrderedSet;
+use crate::pgregression::FloatPairs;
 use crate::pgvariance::{ExactSpread, FloatSpread};
 use crate::quantile::{self, Column, Held, Holistic, Sample};
 use crate::statistics::{Moment, Paired, Pairing, Powers};
@@ -119,6 +120,9 @@ pub(crate) enum General {
     ExactSpread(Box<ExactSpread>),
     /// The variance family of a PostgreSQL session over a double, in [`crate::pgvariance`].
     FloatSpread(FloatSpread),
+    /// `corr`, the covariances and the `regr_*` family of a PostgreSQL session, in
+    /// [`crate::pgregression`].
+    FloatPairs(FloatPairs),
     /// `fsum` and `favg`, a sum with Kahan's running error, in the pin's steps.
     Kahan { value: f64, err: f64, count: u64, average: bool },
     /// `count_if`, the rows that were true, and whether any row was not null.
@@ -225,6 +229,9 @@ impl General {
         if let Some(state) = FloatSpread::named(name) {
             return Some(Self::FloatSpread(state));
         }
+        if let Some(state) = FloatPairs::named(name) {
+            return Some(Self::FloatPairs(state));
+        }
         if name == "avg" {
             return Timed::new(returns).map(Self::Timed);
         }
@@ -320,6 +327,7 @@ impl General {
             Self::Ordered { rows, .. } => rows.push(args.to_vec()),
             Self::Set(state) => state.update(args)?,
             Self::Paired(state) => state.update(args)?,
+            Self::FloatPairs(state) => state.update(args)?,
             _ if value.is_null() => {}
             Self::Counted { key, .. } if args.len() > 1 => {
                 let mut binned = Binned::new(false, key.clone());
@@ -515,14 +523,15 @@ impl General {
     pub(crate) const fn takes_reals(&self, arguments: usize) -> bool {
         matches!(
             (self, arguments),
-            (Self::Paired(_), 2) | (Self::Powers(_) | Self::FloatSpread(_), 1)
+            (Self::Paired(_) | Self::FloatPairs(_), 2) | (Self::Powers(_) | Self::FloatSpread(_), 1)
         )
     }
 
     /// Adds one row of doubles, none of them null, for a state [`Self::takes_reals`] says yes to.
     pub(crate) fn push_reals(&mut self, row: &[f64]) {
         match (self, row) {
             (Self::Paired(state), [y, x]) => state.add(*y, *x),
+            (Self::FloatPairs(state), [y, x]) => state.add(*y, *x),
             (Self::Powers(state), [input]) => state.add(*input),
             (Self::FloatSpread(state), [input]) => state.add(*input),
             _ => {}
@@ -760,6 +769,7 @@ impl General {
             (Self::Powers(state), Self::Powers(theirs)) => state.combine(theirs),
             (Self::ExactSpread(state), Self::ExactSpread(theirs)) => state.combine(theirs)?,
             (Self::FloatSpread(state), Self::FloatSpread(theirs)) => state.combine(theirs)?,
+            (Self::FloatPairs(state), Self::FloatPairs(theirs)) => state.combine(theirs)?,
             (
                 Self::Kahan { value, err, count, .. },
                 Self::Kahan { value: theirs, err: their_err, count: more, .. },
@@ -910,6 +920,7 @@ impl General {
             Self::Powers(state) => state.finish()?,
             Self::ExactSpread(state) => state.finish()?,
             Self::FloatSpread(state) => state.finish()?,
+            Self::FloatPairs(state) => state.finish()?,
             Self::Joined { seen: false, .. } => Value::Null,
             Self::Joined { text, .. } => Value::Varchar(text.clone()),
             Self::Ordered { .. }

crates/rudb-kernels/src/lib.rs

@@ -105,6 +105,7 @@ pub mod pgproc;
 pub mod pgregexp;
 mod pgstring;
 mod pgunicode;
+mod pgregression;
 mod pgvariance;
 pub mod prepare;
 mod printf;