use crate::matrix::common_io::{file_ext, write_lines};
use crate::matrix::parquet::{
parquet_add_numeric_column, parquet_add_string_column, ParquetWriter,
};
use crate::matrix::traits::RunningStatOps;
use nalgebra_sparse::CscMatrix;
use num_traits::{Float, ToPrimitive, Zero};
use parquet::basic::Type as ParquetType;
use std::fmt::Display;
use std::iter::Sum;
use std::ops::AddAssign;
const STAT_COLUMN_NAMES: [&str; 4] = ["nnz", "tot", "mu", "sig"];
fn safe_denom<T: Float>(n: usize) -> T {
let n = T::from(n).unwrap_or(T::one());
if n > T::zero() {
n
} else {
T::from(1e-8).unwrap_or(T::one())
}
}
#[derive(Clone)]
pub struct SparseRunningStatistics<T>
where
T: Float,
{
nrows: usize,
ncols_processed: usize,
npos: Vec<T>,
s1: Vec<T>,
s2: Vec<T>,
}
impl<T> SparseRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero,
{
pub fn new(nrows: usize) -> Self {
SparseRunningStatistics {
nrows,
ncols_processed: 0,
npos: vec![T::zero(); nrows],
s1: vec![T::zero(); nrows],
s2: vec![T::zero(); nrows],
}
}
pub fn add_sparse_column(&mut self, row_indices: &[usize], values: &[T]) {
debug_assert_eq!(row_indices.len(), values.len());
for (&row, &val) in row_indices.iter().zip(values.iter()) {
if val.is_finite() {
if val > T::zero() {
self.npos[row] += T::one();
}
self.s1[row] += val;
self.s2[row] += val * val;
}
}
self.ncols_processed += 1;
}
pub fn nrows(&self) -> usize {
self.nrows
}
pub fn ncols_processed(&self) -> usize {
self.ncols_processed
}
fn denom(&self) -> T {
safe_denom::<T>(self.ncols_processed)
}
pub fn to_vecs(&self) -> (Vec<T>, Vec<T>, Vec<T>, Vec<T>) {
(self.npos.clone(), self.s1.clone(), self.mean(), self.std())
}
pub fn add_csc(&mut self, csc: &CscMatrix<T>) {
for col in csc.col_iter() {
let rows = col.row_indices();
let vals = col.values();
self.add_sparse_column(rows, vals);
}
}
pub fn add_dense_column(&mut self, values: &[T]) {
debug_assert_eq!(values.len(), self.nrows);
let zero = T::zero();
let one = T::one();
for ((v_in, npos), (s1, s2)) in values
.iter()
.zip(self.npos.iter_mut())
.zip(self.s1.iter_mut().zip(self.s2.iter_mut()))
{
let v = *v_in;
let v = if v.is_finite() { v } else { zero };
let pos = if v > zero { one } else { zero };
*npos += pos;
*s1 += v;
*s2 += v * v;
}
self.ncols_processed += 1;
}
pub fn add_dense_column_scaled(&mut self, values: &[T], scale: T) {
debug_assert_eq!(values.len(), self.nrows);
let zero = T::zero();
let one = T::one();
for ((v_in, npos), (s1, s2)) in values
.iter()
.zip(self.npos.iter_mut())
.zip(self.s1.iter_mut().zip(self.s2.iter_mut()))
{
let v = *v_in * scale;
let v = if v.is_finite() { v } else { zero };
let pos = if v > zero { one } else { zero };
*npos += pos;
*s1 += v;
*s2 += v * v;
}
self.ncols_processed += 1;
}
pub fn add_dense_columns(&mut self, dense: &nalgebra::DMatrix<T>)
where
T: nalgebra::Scalar,
{
debug_assert_eq!(dense.nrows(), self.nrows);
for j in 0..dense.ncols() {
let col = dense.column(j);
self.add_dense_column(col.as_slice());
}
}
pub fn merge(&mut self, other: &Self) {
debug_assert_eq!(self.nrows, other.nrows);
for (a, b) in self.npos.iter_mut().zip(other.npos.iter()) {
*a += *b;
}
for (a, b) in self.s1.iter_mut().zip(other.s1.iter()) {
*a += *b;
}
for (a, b) in self.s2.iter_mut().zip(other.s2.iter()) {
*a += *b;
}
self.ncols_processed += other.ncols_processed;
}
}
impl<T> SparseRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero + Display + ToPrimitive,
{
pub fn save(&self, filename: &str, names: &[Box<str>], sep: &str) -> anyhow::Result<()> {
let (nnz, tot, mu, sig) = self.to_f32_vecs();
write_stat_file(
filename,
names,
sep,
StatColumns {
nnz: &nnz,
tot: &tot,
mu: &mu,
sig: &sig,
},
)
}
pub fn to_f32_vecs(&self) -> (Vec<f32>, Vec<f32>, Vec<f32>, Vec<f32>) {
let to_f32_slice =
|v: &[T]| -> Vec<f32> { v.iter().map(|x| x.to_f32().unwrap_or(0.0)).collect() };
let nnz = to_f32_slice(&self.npos);
let tot = to_f32_slice(&self.s1);
let mu = to_f32_slice(&self.mean());
let sig = to_f32_slice(&self.std());
(nnz, tot, mu, sig)
}
pub fn to_string_vec(&self, names: &[Box<str>], sep: &str) -> anyhow::Result<Vec<Box<str>>> {
if names.len() != self.nrows {
anyhow::bail!(
"The number of names ({}) does not match nrows ({})",
names.len(),
self.nrows
);
}
let nnz = self.count_positives();
let tot = self.sum();
let mu = self.mean();
let sig = self.std();
let out: Vec<Box<str>> = (0..self.nrows)
.map(|i| {
format!(
"{}{}{}{}{}{}{}{}{}",
names[i],
sep,
format_value(nnz[i]),
sep,
format_value(tot[i]),
sep,
format_value(mu[i]),
sep,
format_value(sig[i])
)
.into_boxed_str()
})
.collect();
Ok(out)
}
}
#[derive(Clone)]
pub struct SparseColumnRunningStatistics<T>
where
T: Float,
{
nrows: usize,
npos: Vec<T>,
s1: Vec<T>,
s2: Vec<T>,
}
impl<T> SparseColumnRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero,
{
pub fn new(ncols: usize, nrows: usize) -> Self {
Self {
nrows,
npos: vec![T::zero(); ncols],
s1: vec![T::zero(); ncols],
s2: vec![T::zero(); ncols],
}
}
pub fn add_csc(&mut self, csc: &CscMatrix<T>, col_offset: usize) {
self.add_csc_inner(csc, col_offset, None);
}
pub fn add_csc_masked(&mut self, csc: &CscMatrix<T>, col_offset: usize, row_mask: &[bool]) {
debug_assert!(row_mask.len() >= csc.nrows());
self.add_csc_inner(csc, col_offset, Some(row_mask));
}
fn add_csc_inner(&mut self, csc: &CscMatrix<T>, col_offset: usize, row_mask: Option<&[bool]>) {
for (local_col, col) in csc.col_iter().enumerate() {
let c = col_offset + local_col;
let rows = col.row_indices();
let vals = col.values();
for (&row, &val) in rows.iter().zip(vals.iter()) {
if let Some(mask) = row_mask {
if !mask[row] {
continue;
}
}
if !val.is_finite() {
continue;
}
if val > T::zero() {
self.npos[c] += T::one();
}
self.s1[c] += val;
self.s2[c] += val * val;
}
}
}
pub fn ncols(&self) -> usize {
self.npos.len()
}
pub fn nrows(&self) -> usize {
self.nrows
}
fn denom(&self) -> T {
safe_denom::<T>(self.nrows)
}
}
impl<T> SparseColumnRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero + Display + ToPrimitive,
{
pub fn save(&self, filename: &str, names: &[Box<str>], sep: &str) -> anyhow::Result<()> {
let (nnz, tot, mu, sig) = self.to_f32_vecs();
write_stat_file(
filename,
names,
sep,
StatColumns {
nnz: &nnz,
tot: &tot,
mu: &mu,
sig: &sig,
},
)
}
pub fn to_f32_vecs(&self) -> (Vec<f32>, Vec<f32>, Vec<f32>, Vec<f32>) {
let to_f32_slice =
|v: &[T]| -> Vec<f32> { v.iter().map(|x| x.to_f32().unwrap_or(0.0)).collect() };
let nnz = to_f32_slice(&self.npos);
let tot = to_f32_slice(&self.s1);
let mu = to_f32_slice(&self.mean());
let sig = to_f32_slice(&self.std());
(nnz, tot, mu, sig)
}
}
impl<T> RunningStatOps<T> for SparseColumnRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero,
{
type Output = Vec<T>;
fn clear(&mut self) {
self.npos.fill(T::zero());
self.s1.fill(T::zero());
self.s2.fill(T::zero());
}
fn count_positives(&self) -> Vec<T> {
self.npos.clone()
}
fn sum(&self) -> Vec<T> {
self.s1.clone()
}
fn mean(&self) -> Vec<T> {
let n = self.denom();
self.s1.iter().map(|&s| s / n).collect()
}
fn variance(&self) -> Vec<T> {
let n = self.denom();
self.s1
.iter()
.zip(self.s2.iter())
.map(|(&s1, &s2)| {
let mu = s1 / n;
s2 / n - mu * mu
})
.collect()
}
fn std(&self) -> Vec<T> {
self.variance().into_iter().map(|v| v.sqrt()).collect()
}
}
struct StatColumns<'a> {
nnz: &'a [f32],
tot: &'a [f32],
mu: &'a [f32],
sig: &'a [f32],
}
impl StatColumns<'_> {
fn len(&self) -> usize {
self.nnz.len()
}
}
fn write_stat_file(
filename: &str,
names: &[Box<str>],
sep: &str,
stats: StatColumns<'_>,
) -> anyhow::Result<()> {
let n = stats.len();
if names.len() != n {
anyhow::bail!(
"The number of names ({}) does not match stat length ({})",
names.len(),
n
);
}
match file_ext(filename).unwrap_or(Box::from("")).as_ref() {
"parquet" => {
let column_names: Vec<Box<str>> =
STAT_COLUMN_NAMES.iter().map(|s| (*s).into()).collect();
let column_types = vec![ParquetType::FLOAT; STAT_COLUMN_NAMES.len()];
let parquet_writer = ParquetWriter::new(
filename,
(n, 4),
(Some(names), Some(&column_names)),
Some(&column_types),
None,
)?;
let mut writer = parquet_writer.get_writer()?;
let mut row_group_writer = writer.next_row_group()?;
parquet_add_string_column(&mut row_group_writer, names)?;
parquet_add_numeric_column(&mut row_group_writer, stats.nnz)?;
parquet_add_numeric_column(&mut row_group_writer, stats.tot)?;
parquet_add_numeric_column(&mut row_group_writer, stats.mu)?;
parquet_add_numeric_column(&mut row_group_writer, stats.sig)?;
row_group_writer.close()?;
writer.close()?;
}
_ => {
let mut out: Vec<Box<str>> = (0..n)
.map(|i| {
format!(
"{}{}{}{}{}{}{}{}{}",
names[i],
sep,
format_value(stats.nnz[i]),
sep,
format_value(stats.tot[i]),
sep,
format_value(stats.mu[i]),
sep,
format_value(stats.sig[i])
)
.into_boxed_str()
})
.collect();
let header = format!("#name{}nnz{}tot{}mu{}sig", sep, sep, sep, sep);
out.insert(0, header.into_boxed_str());
write_lines(&out, filename)?;
}
}
Ok(())
}
pub fn save_grouped_stats_parquet(
filename: &str,
names: &[Box<str>],
group_names: &[Box<str>],
group_stats: &[SparseRunningStatistics<f32>],
) -> anyhow::Result<()> {
save_grouped_stats_parquet_cols(filename, &[("name", names)], group_names, group_stats)
}
pub fn save_grouped_stats_parquet_cols(
filename: &str,
key_cols: &[(&str, &[Box<str>])],
group_names: &[Box<str>],
group_stats: &[SparseRunningStatistics<f32>],
) -> anyhow::Result<()> {
use crate::matrix::parquet::{write_named_table, Column};
if group_names.len() != group_stats.len() {
anyhow::bail!(
"Number of group names ({}) does not match number of group stats ({})",
group_names.len(),
group_stats.len()
);
}
anyhow::ensure!(
!key_cols.is_empty(),
"save_grouped_stats_parquet_cols: need at least one key column"
);
let n_features = group_stats.first().map_or(0, |s| s.nrows());
for &(name, vals) in key_cols {
anyhow::ensure!(
vals.len() == n_features,
"key column '{name}' has {} entries but there are {n_features} feature rows",
vals.len(),
);
}
let total_rows = n_features * group_names.len();
let mut keys: Vec<Vec<Box<str>>> = (0..key_cols.len())
.map(|_| Vec::with_capacity(total_rows))
.collect();
let mut all_groups: Vec<Box<str>> = Vec::with_capacity(total_rows);
let mut all_nnz: Vec<f32> = Vec::with_capacity(total_rows);
let mut all_tot: Vec<f32> = Vec::with_capacity(total_rows);
let mut all_mu: Vec<f32> = Vec::with_capacity(total_rows);
let mut all_sig: Vec<f32> = Vec::with_capacity(total_rows);
for (group_name, stat) in group_names.iter().zip(group_stats.iter()) {
let (nnz, tot, mu, sig) = stat.to_f32_vecs();
for (j, &(_, vals)) in key_cols.iter().enumerate() {
keys[j].extend(vals.iter().cloned());
}
all_groups.resize(all_groups.len() + n_features, group_name.clone());
all_nnz.extend_from_slice(&nnz);
all_tot.extend_from_slice(&tot);
all_mu.extend_from_slice(&mu);
all_sig.extend_from_slice(&sig);
}
let mut columns: Vec<(Box<str>, Column)> = Vec::with_capacity(key_cols.len() + 4);
for (&(name, _), col) in key_cols.iter().zip(keys.iter()).skip(1) {
columns.push((name.into(), Column::Str(col.as_slice())));
}
columns.push(("group".into(), Column::Str(all_groups.as_slice())));
columns.push(("nnz".into(), Column::F32(all_nnz.as_slice())));
columns.push(("tot".into(), Column::F32(all_tot.as_slice())));
columns.push(("mu".into(), Column::F32(all_mu.as_slice())));
columns.push(("sig".into(), Column::F32(all_sig.as_slice())));
write_named_table(filename, key_cols[0].0, &keys[0], &columns)
}
fn format_value<T: Float + Display>(v: T) -> String {
let v_f64 = v.to_f64().unwrap_or(0.0);
if v_f64.abs() > 1e-4 {
format!("{:.4}", v_f64)
.trim_end_matches('0')
.trim_end_matches('.')
.to_string()
} else if v_f64.abs() > 1e-20 {
format!("{:.4e}", v_f64)
} else {
"0".to_string()
}
}
impl<T> RunningStatOps<T> for SparseRunningStatistics<T>
where
T: Float + AddAssign + Sum + Zero,
{
type Output = Vec<T>;
fn clear(&mut self) {
self.ncols_processed = 0;
self.npos.fill(T::zero());
self.s1.fill(T::zero());
self.s2.fill(T::zero());
}
fn count_positives(&self) -> Vec<T> {
self.npos.clone()
}
fn sum(&self) -> Vec<T> {
self.s1.clone()
}
fn mean(&self) -> Vec<T> {
let n = self.denom();
self.s1.iter().map(|&s| s / n).collect()
}
fn variance(&self) -> Vec<T> {
let n = self.denom();
self.s1
.iter()
.zip(self.s2.iter())
.map(|(&s1, &s2)| {
let mu = s1 / n;
s2 / n - mu * mu
})
.collect()
}
fn std(&self) -> Vec<T> {
self.variance().into_iter().map(|v| v.sqrt()).collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_sparse_running_stat_basic() {
let mut stat = SparseRunningStatistics::<f32>::new(4);
stat.add_sparse_column(&[0, 2], &[1.0, 2.0]);
stat.add_sparse_column(&[1, 3], &[3.0, 4.0]);
assert_eq!(stat.ncols_processed(), 2);
assert_eq!(stat.count_positives(), vec![1.0, 1.0, 1.0, 1.0]);
assert_eq!(stat.sum(), vec![1.0, 3.0, 2.0, 4.0]);
let mean = stat.mean();
assert!((mean[0] - 0.5).abs() < 1e-6);
assert!((mean[1] - 1.5).abs() < 1e-6);
assert!((mean[2] - 1.0).abs() < 1e-6);
assert!((mean[3] - 2.0).abs() < 1e-6);
}
#[test]
fn test_sparse_running_stat_csc() {
use nalgebra_sparse::CooMatrix;
let mut stat = SparseRunningStatistics::<f32>::new(3);
let mut coo: CooMatrix<f32> = CooMatrix::new(3, 2);
coo.push(0, 0, 1.0);
coo.push(1, 1, 2.0);
coo.push(2, 0, 3.0);
let csc = CscMatrix::from(&coo);
stat.add_csc(&csc);
assert_eq!(stat.ncols_processed(), 2);
assert_eq!(stat.count_positives(), vec![1.0, 1.0, 1.0]);
assert_eq!(stat.sum(), vec![1.0, 2.0, 3.0]);
}
#[test]
fn test_sparse_running_stat_f64() {
let mut stat = SparseRunningStatistics::<f64>::new(2);
stat.add_sparse_column(&[0, 1], &[1.0, 2.0]);
stat.add_sparse_column(&[0], &[3.0]);
assert_eq!(stat.ncols_processed(), 2);
assert_eq!(stat.sum(), vec![4.0, 2.0]);
let mean = stat.mean();
assert!((mean[0] - 2.0).abs() < 1e-10);
assert!((mean[1] - 1.0).abs() < 1e-10);
}
#[test]
fn test_sparse_column_running_stat_csc() {
use nalgebra_sparse::CooMatrix;
let mut coo: CooMatrix<f32> = CooMatrix::new(3, 4);
coo.push(0, 0, 1.0);
coo.push(2, 0, 3.0);
coo.push(1, 1, 2.0);
coo.push(0, 3, 4.0);
coo.push(1, 3, 5.0);
let csc = CscMatrix::from(&coo);
let mut stat = SparseColumnRunningStatistics::<f32>::new(4, 3);
stat.add_csc(&csc, 0);
assert_eq!(stat.count_positives(), vec![2.0, 1.0, 0.0, 2.0]);
assert_eq!(stat.sum(), vec![4.0, 2.0, 0.0, 9.0]);
let mean = stat.mean();
assert!((mean[0] - 4.0 / 3.0).abs() < 1e-6);
assert!((mean[1] - 2.0 / 3.0).abs() < 1e-6);
assert!((mean[2] - 0.0).abs() < 1e-6);
assert!((mean[3] - 9.0 / 3.0).abs() < 1e-6);
}
#[test]
fn test_sparse_column_running_stat_block_offset() {
use nalgebra_sparse::CooMatrix;
let mut coo_a: CooMatrix<f32> = CooMatrix::new(3, 2);
coo_a.push(0, 0, 1.0);
coo_a.push(2, 0, 3.0);
coo_a.push(1, 1, 2.0);
let csc_a = CscMatrix::from(&coo_a);
let mut coo_b: CooMatrix<f32> = CooMatrix::new(3, 2);
coo_b.push(0, 1, 4.0);
coo_b.push(1, 1, 5.0);
let csc_b = CscMatrix::from(&coo_b);
let mut stat = SparseColumnRunningStatistics::<f32>::new(4, 3);
stat.add_csc(&csc_a, 0);
stat.add_csc(&csc_b, 2);
assert_eq!(stat.count_positives(), vec![2.0, 1.0, 0.0, 2.0]);
assert_eq!(stat.sum(), vec![4.0, 2.0, 0.0, 9.0]);
}
#[test]
fn test_sparse_column_running_stat_masked() {
use nalgebra_sparse::CooMatrix;
let mut coo: CooMatrix<f32> = CooMatrix::new(4, 2);
coo.push(0, 0, 1.0);
coo.push(1, 0, 2.0);
coo.push(2, 0, 3.0);
coo.push(3, 0, 4.0);
coo.push(1, 1, 5.0);
coo.push(3, 1, 6.0);
let csc = CscMatrix::from(&coo);
let row_mask = vec![true, false, true, false];
let mut stat = SparseColumnRunningStatistics::<f32>::new(2, 2);
stat.add_csc_masked(&csc, 0, &row_mask);
assert_eq!(stat.count_positives(), vec![2.0, 0.0]);
assert_eq!(stat.sum(), vec![4.0, 0.0]);
}
}