| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| //! [`VarianceSample`]: variance sample aggregations. |
| //! [`VariancePopulation`]: variance population aggregations. |
| |
| use arrow::datatypes::{FieldRef, Float64Type}; |
| use arrow::{ |
| array::{Array, ArrayRef, BooleanArray, Float64Array, UInt64Array}, |
| buffer::NullBuffer, |
| datatypes::{DataType, Field}, |
| }; |
| use datafusion_common::cast::{as_float64_array, as_uint64_array}; |
| use datafusion_common::{Result, ScalarValue}; |
| use datafusion_expr::{ |
| Accumulator, AggregateUDFImpl, Documentation, GroupsAccumulator, Signature, |
| Volatility, |
| function::{AccumulatorArgs, StateFieldsArgs}, |
| utils::format_state_name, |
| }; |
| use datafusion_functions_aggregate_common::utils::GenericDistinctBuffer; |
| use datafusion_functions_aggregate_common::{ |
| aggregate::groups_accumulator::accumulate::accumulate, stats::StatsType, |
| }; |
| use datafusion_macros::user_doc; |
| use std::mem::{size_of, size_of_val}; |
| use std::{fmt::Debug, sync::Arc}; |
| |
| make_udaf_expr_and_func!( |
| VarianceSample, |
| var_sample, |
| expression, |
| "Computes the sample variance.", |
| var_samp_udaf |
| ); |
| |
| make_udaf_expr_and_func!( |
| VariancePopulation, |
| var_pop, |
| expression, |
| "Computes the population variance.", |
| var_pop_udaf |
| ); |
| |
| #[user_doc( |
| doc_section(label = "General Functions"), |
| description = "Returns the statistical sample variance of a set of numbers.", |
| syntax_example = "var(expression)", |
| standard_argument(name = "expression", prefix = "Numeric") |
| )] |
| #[derive(PartialEq, Eq, Hash, Debug)] |
| pub struct VarianceSample { |
| signature: Signature, |
| aliases: Vec<String>, |
| } |
| |
| impl Default for VarianceSample { |
| fn default() -> Self { |
| Self::new() |
| } |
| } |
| |
| impl VarianceSample { |
| pub fn new() -> Self { |
| Self { |
| aliases: vec![String::from("var_sample"), String::from("var_samp")], |
| signature: Signature::exact(vec![DataType::Float64], Volatility::Immutable), |
| } |
| } |
| } |
| |
| impl AggregateUDFImpl for VarianceSample { |
| fn name(&self) -> &str { |
| "var" |
| } |
| |
| fn signature(&self) -> &Signature { |
| &self.signature |
| } |
| |
| fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> { |
| Ok(DataType::Float64) |
| } |
| |
| fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> { |
| let name = args.name; |
| match args.is_distinct { |
| false => Ok(vec![ |
| Field::new(format_state_name(name, "count"), DataType::UInt64, true), |
| Field::new(format_state_name(name, "mean"), DataType::Float64, true), |
| Field::new(format_state_name(name, "m2"), DataType::Float64, true), |
| ] |
| .into_iter() |
| .map(Arc::new) |
| .collect()), |
| true => { |
| let field = Field::new_list_field(DataType::Float64, true); |
| let state_name = "distinct_var"; |
| Ok(vec![ |
| Field::new( |
| format_state_name(name, state_name), |
| DataType::List(Arc::new(field)), |
| true, |
| ) |
| .into(), |
| ]) |
| } |
| } |
| } |
| |
| fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { |
| if acc_args.is_distinct { |
| return Ok(Box::new(DistinctVarianceAccumulator::new( |
| StatsType::Sample, |
| ))); |
| } |
| |
| Ok(Box::new(VarianceAccumulator::try_new(StatsType::Sample)?)) |
| } |
| |
| fn aliases(&self) -> &[String] { |
| &self.aliases |
| } |
| |
| fn groups_accumulator_supported(&self, acc_args: AccumulatorArgs) -> bool { |
| !acc_args.is_distinct |
| } |
| |
| fn create_groups_accumulator( |
| &self, |
| _args: AccumulatorArgs, |
| ) -> Result<Box<dyn GroupsAccumulator>> { |
| Ok(Box::new(VarianceGroupsAccumulator::new(StatsType::Sample))) |
| } |
| |
| fn documentation(&self) -> Option<&Documentation> { |
| self.doc() |
| } |
| } |
| |
| #[user_doc( |
| doc_section(label = "General Functions"), |
| description = "Returns the statistical population variance of a set of numbers.", |
| syntax_example = "var_pop(expression)", |
| standard_argument(name = "expression", prefix = "Numeric") |
| )] |
| #[derive(PartialEq, Eq, Hash, Debug)] |
| pub struct VariancePopulation { |
| signature: Signature, |
| aliases: Vec<String>, |
| } |
| |
| impl Default for VariancePopulation { |
| fn default() -> Self { |
| Self::new() |
| } |
| } |
| |
| impl VariancePopulation { |
| pub fn new() -> Self { |
| Self { |
| aliases: vec![String::from("var_population")], |
| signature: Signature::exact(vec![DataType::Float64], Volatility::Immutable), |
| } |
| } |
| } |
| |
| impl AggregateUDFImpl for VariancePopulation { |
| fn name(&self) -> &str { |
| "var_pop" |
| } |
| |
| fn signature(&self) -> &Signature { |
| &self.signature |
| } |
| |
| fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> { |
| Ok(DataType::Float64) |
| } |
| |
| fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> { |
| match args.is_distinct { |
| false => { |
| let name = args.name; |
| Ok(vec![ |
| Field::new(format_state_name(name, "count"), DataType::UInt64, true), |
| Field::new(format_state_name(name, "mean"), DataType::Float64, true), |
| Field::new(format_state_name(name, "m2"), DataType::Float64, true), |
| ] |
| .into_iter() |
| .map(Arc::new) |
| .collect()) |
| } |
| true => { |
| let field = Field::new_list_field(DataType::Float64, true); |
| let state_name = "distinct_var"; |
| Ok(vec![ |
| Field::new( |
| format_state_name(args.name, state_name), |
| DataType::List(Arc::new(field)), |
| true, |
| ) |
| .into(), |
| ]) |
| } |
| } |
| } |
| |
| fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { |
| if acc_args.is_distinct { |
| return Ok(Box::new(DistinctVarianceAccumulator::new( |
| StatsType::Population, |
| ))); |
| } |
| |
| Ok(Box::new(VarianceAccumulator::try_new( |
| StatsType::Population, |
| )?)) |
| } |
| |
| fn aliases(&self) -> &[String] { |
| &self.aliases |
| } |
| |
| fn groups_accumulator_supported(&self, acc_args: AccumulatorArgs) -> bool { |
| !acc_args.is_distinct |
| } |
| |
| fn create_groups_accumulator( |
| &self, |
| _args: AccumulatorArgs, |
| ) -> Result<Box<dyn GroupsAccumulator>> { |
| Ok(Box::new(VarianceGroupsAccumulator::new( |
| StatsType::Population, |
| ))) |
| } |
| |
| fn documentation(&self) -> Option<&Documentation> { |
| self.doc() |
| } |
| } |
| |
| /// An accumulator to compute variance |
| /// The algorithm used is an online implementation and numerically stable. It is based on this paper: |
| /// Welford, B. P. (1962). "Note on a method for calculating corrected sums of squares and products". |
| /// Technometrics. 4 (3): 419–420. doi:10.2307/1266577. JSTOR 1266577. |
| /// |
| /// The algorithm has been analyzed here: |
| /// Ling, Robert F. (1974). "Comparison of Several Algorithms for Computing Sample Means and Variances". |
| /// Journal of the American Statistical Association. 69 (348): 859–866. doi:10.2307/2286154. JSTOR 2286154. |
| |
| #[derive(Debug)] |
| pub struct VarianceAccumulator { |
| m2: f64, |
| mean: f64, |
| count: u64, |
| stats_type: StatsType, |
| } |
| |
| impl VarianceAccumulator { |
| /// Creates a new `VarianceAccumulator` |
| pub fn try_new(s_type: StatsType) -> Result<Self> { |
| Ok(Self { |
| m2: 0_f64, |
| mean: 0_f64, |
| count: 0_u64, |
| stats_type: s_type, |
| }) |
| } |
| |
| pub fn get_count(&self) -> u64 { |
| self.count |
| } |
| |
| pub fn get_mean(&self) -> f64 { |
| self.mean |
| } |
| |
| pub fn get_m2(&self) -> f64 { |
| self.m2 |
| } |
| } |
| |
| #[inline] |
| fn merge( |
| count: u64, |
| mean: f64, |
| m2: f64, |
| count2: u64, |
| mean2: f64, |
| m22: f64, |
| ) -> (u64, f64, f64) { |
| debug_assert!(count != 0 || count2 != 0, "Cannot merge two empty states"); |
| let new_count = count + count2; |
| let new_mean = |
| mean * count as f64 / new_count as f64 + mean2 * count2 as f64 / new_count as f64; |
| let delta = mean - mean2; |
| let new_m2 = |
| m2 + m22 + delta * delta * count as f64 * count2 as f64 / new_count as f64; |
| |
| (new_count, new_mean, new_m2) |
| } |
| |
| #[inline] |
| fn update(count: u64, mean: f64, m2: f64, value: f64) -> (u64, f64, f64) { |
| let new_count = count + 1; |
| let delta1 = value - mean; |
| let new_mean = delta1 / new_count as f64 + mean; |
| let delta2 = value - new_mean; |
| let new_m2 = m2 + delta1 * delta2; |
| |
| (new_count, new_mean, new_m2) |
| } |
| |
| /// Inverse of [`update`]: removes a previously accumulated value. Retracting |
| /// from a state with one or zero values resets the state to empty. |
| #[inline] |
| fn retract(count: u64, mean: f64, m2: f64, value: f64) -> (u64, f64, f64) { |
| if count <= 1 { |
| return (0, 0.0, 0.0); |
| } |
| |
| let new_count = count - 1; |
| let delta1 = mean - value; |
| let new_mean = delta1 / new_count as f64 + mean; |
| let delta2 = new_mean - value; |
| let new_m2 = m2 - delta1 * delta2; |
| |
| (new_count, new_mean, new_m2) |
| } |
| |
| impl Accumulator for VarianceAccumulator { |
| fn state(&mut self) -> Result<Vec<ScalarValue>> { |
| Ok(vec![ |
| ScalarValue::from(self.count), |
| ScalarValue::from(self.mean), |
| ScalarValue::from(self.m2), |
| ]) |
| } |
| |
| fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { |
| let arr = as_float64_array(&values[0])?; |
| for value in arr.iter().flatten() { |
| (self.count, self.mean, self.m2) = |
| update(self.count, self.mean, self.m2, value) |
| } |
| |
| Ok(()) |
| } |
| |
| fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> { |
| let arr = as_float64_array(&values[0])?; |
| for value in arr.iter().flatten() { |
| (self.count, self.mean, self.m2) = |
| retract(self.count, self.mean, self.m2, value) |
| } |
| |
| Ok(()) |
| } |
| |
| fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { |
| let counts = as_uint64_array(&states[0])?; |
| let means = as_float64_array(&states[1])?; |
| let m2s = as_float64_array(&states[2])?; |
| |
| for i in 0..counts.len() { |
| let c = counts.value(i); |
| if c == 0_u64 { |
| continue; |
| } |
| (self.count, self.mean, self.m2) = merge( |
| self.count, |
| self.mean, |
| self.m2, |
| c, |
| means.value(i), |
| m2s.value(i), |
| ) |
| } |
| Ok(()) |
| } |
| |
| fn evaluate(&mut self) -> Result<ScalarValue> { |
| let count = match self.stats_type { |
| StatsType::Population => self.count, |
| StatsType::Sample => { |
| if self.count > 0 { |
| self.count - 1 |
| } else { |
| self.count |
| } |
| } |
| }; |
| |
| Ok(ScalarValue::Float64(match self.count { |
| 0 => None, |
| 1 => { |
| if let StatsType::Population = self.stats_type { |
| Some(0.0) |
| } else { |
| None |
| } |
| } |
| _ => Some(self.m2 / count as f64), |
| })) |
| } |
| |
| fn size(&self) -> usize { |
| size_of_val(self) |
| } |
| |
| fn supports_retract_batch(&self) -> bool { |
| true |
| } |
| } |
| |
| #[derive(Debug)] |
| pub struct VarianceGroupsAccumulator { |
| m2s: Vec<f64>, |
| means: Vec<f64>, |
| counts: Vec<u64>, |
| stats_type: StatsType, |
| } |
| |
| impl VarianceGroupsAccumulator { |
| pub fn new(s_type: StatsType) -> Self { |
| Self { |
| m2s: Vec::new(), |
| means: Vec::new(), |
| counts: Vec::new(), |
| stats_type: s_type, |
| } |
| } |
| |
| fn resize(&mut self, total_num_groups: usize) { |
| self.m2s.resize(total_num_groups, 0.0); |
| self.means.resize(total_num_groups, 0.0); |
| self.counts.resize(total_num_groups, 0); |
| } |
| |
| fn merge<F>( |
| group_indices: &[usize], |
| counts: &UInt64Array, |
| means: &Float64Array, |
| m2s: &Float64Array, |
| _opt_filter: Option<&BooleanArray>, |
| mut value_fn: F, |
| ) where |
| F: FnMut(usize, u64, f64, f64) + Send, |
| { |
| assert_eq!(counts.null_count(), 0); |
| assert_eq!(means.null_count(), 0); |
| assert_eq!(m2s.null_count(), 0); |
| |
| group_indices |
| .iter() |
| .zip(counts.values().iter()) |
| .zip(means.values().iter()) |
| .zip(m2s.values().iter()) |
| .for_each(|(((&group_index, &count), &mean), &m2)| { |
| value_fn(group_index, count, mean, m2); |
| }); |
| } |
| |
| pub fn variance( |
| &mut self, |
| emit_to: datafusion_expr::EmitTo, |
| ) -> (Vec<f64>, NullBuffer) { |
| let mut counts = emit_to.take_needed(&mut self.counts); |
| // means are only needed for updating m2s and are not needed for the final result. |
| // But we still need to take them to ensure the internal state is consistent. |
| let _ = emit_to.take_needed(&mut self.means); |
| let m2s = emit_to.take_needed(&mut self.m2s); |
| |
| if let StatsType::Sample = self.stats_type { |
| counts.iter_mut().for_each(|count| { |
| *count = count.saturating_sub(1); |
| }); |
| } |
| let nulls = NullBuffer::from_iter(counts.iter().map(|&count| count != 0)); |
| let variance = m2s |
| .iter() |
| .zip(counts) |
| .map(|(m2, count)| m2 / count as f64) |
| .collect(); |
| (variance, nulls) |
| } |
| } |
| |
| impl GroupsAccumulator for VarianceGroupsAccumulator { |
| fn update_batch( |
| &mut self, |
| values: &[ArrayRef], |
| group_indices: &[usize], |
| opt_filter: Option<&BooleanArray>, |
| total_num_groups: usize, |
| ) -> Result<()> { |
| assert_eq!(values.len(), 1, "single argument to update_batch"); |
| let values = as_float64_array(&values[0])?; |
| |
| self.resize(total_num_groups); |
| accumulate(group_indices, values, opt_filter, |group_index, value| { |
| let (new_count, new_mean, new_m2) = update( |
| self.counts[group_index], |
| self.means[group_index], |
| self.m2s[group_index], |
| value, |
| ); |
| self.counts[group_index] = new_count; |
| self.means[group_index] = new_mean; |
| self.m2s[group_index] = new_m2; |
| }); |
| Ok(()) |
| } |
| |
| fn merge_batch( |
| &mut self, |
| values: &[ArrayRef], |
| group_indices: &[usize], |
| total_num_groups: usize, |
| ) -> Result<()> { |
| assert_eq!(values.len(), 3, "two arguments to merge_batch"); |
| // first batch is counts, second is partial means, third is partial m2s |
| let partial_counts = as_uint64_array(&values[0])?; |
| let partial_means = as_float64_array(&values[1])?; |
| let partial_m2s = as_float64_array(&values[2])?; |
| |
| self.resize(total_num_groups); |
| Self::merge( |
| group_indices, |
| partial_counts, |
| partial_means, |
| partial_m2s, |
| None, |
| |group_index, partial_count, partial_mean, partial_m2| { |
| if partial_count == 0 { |
| return; |
| } |
| let (new_count, new_mean, new_m2) = merge( |
| self.counts[group_index], |
| self.means[group_index], |
| self.m2s[group_index], |
| partial_count, |
| partial_mean, |
| partial_m2, |
| ); |
| self.counts[group_index] = new_count; |
| self.means[group_index] = new_mean; |
| self.m2s[group_index] = new_m2; |
| }, |
| ); |
| Ok(()) |
| } |
| |
| fn evaluate(&mut self, emit_to: datafusion_expr::EmitTo) -> Result<ArrayRef> { |
| let (variances, nulls) = self.variance(emit_to); |
| Ok(Arc::new(Float64Array::new(variances.into(), Some(nulls)))) |
| } |
| |
| fn state(&mut self, emit_to: datafusion_expr::EmitTo) -> Result<Vec<ArrayRef>> { |
| let counts = emit_to.take_needed(&mut self.counts); |
| let means = emit_to.take_needed(&mut self.means); |
| let m2s = emit_to.take_needed(&mut self.m2s); |
| |
| Ok(vec![ |
| Arc::new(UInt64Array::new(counts.into(), None)), |
| Arc::new(Float64Array::new(means.into(), None)), |
| Arc::new(Float64Array::new(m2s.into(), None)), |
| ]) |
| } |
| |
| fn convert_to_state( |
| &self, |
| values: &[ArrayRef], |
| opt_filter: Option<&BooleanArray>, |
| ) -> Result<Vec<ArrayRef>> { |
| assert_eq!(values.len(), 1, "single argument to convert_to_state"); |
| let values = as_float64_array(&values[0])?; |
| |
| let len = values.len(); |
| let mut counts = Vec::with_capacity(len); |
| let mut means = Vec::with_capacity(len); |
| let mut m2s = Vec::with_capacity(len); |
| |
| for row in 0..len { |
| if values.is_valid(row) |
| && opt_filter |
| .is_none_or(|filter| filter.is_valid(row) && filter.value(row)) |
| { |
| counts.push(1); |
| means.push(values.value(row)); |
| } else { |
| counts.push(0); |
| means.push(0.0); |
| } |
| m2s.push(0.0); |
| } |
| |
| Ok(vec![ |
| Arc::new(UInt64Array::new(counts.into(), None)), |
| Arc::new(Float64Array::new(means.into(), None)), |
| Arc::new(Float64Array::new(m2s.into(), None)), |
| ]) |
| } |
| fn size(&self) -> usize { |
| self.m2s.capacity() * size_of::<f64>() |
| + self.means.capacity() * size_of::<f64>() |
| + self.counts.capacity() * size_of::<u64>() |
| } |
| } |
| |
| #[derive(Debug)] |
| pub struct DistinctVarianceAccumulator { |
| distinct_values: GenericDistinctBuffer<Float64Type>, |
| stat_type: StatsType, |
| } |
| |
| impl DistinctVarianceAccumulator { |
| pub fn new(stat_type: StatsType) -> Self { |
| Self { |
| distinct_values: GenericDistinctBuffer::<Float64Type>::new(DataType::Float64), |
| stat_type, |
| } |
| } |
| } |
| |
| impl Accumulator for DistinctVarianceAccumulator { |
| fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> { |
| self.distinct_values.update_batch(values) |
| } |
| |
| fn evaluate(&mut self) -> Result<ScalarValue> { |
| let values = self |
| .distinct_values |
| .values |
| .iter() |
| .map(|v| v.0) |
| .collect::<Vec<_>>(); |
| |
| let count = match self.stat_type { |
| StatsType::Sample => { |
| if !values.is_empty() { |
| values.len() - 1 |
| } else { |
| 0 |
| } |
| } |
| StatsType::Population => values.len(), |
| }; |
| |
| let mean = values.iter().sum::<f64>() / values.len() as f64; |
| let m2 = values.iter().map(|x| (x - mean) * (x - mean)).sum::<f64>(); |
| |
| Ok(ScalarValue::Float64(match values.len() { |
| 0 => None, |
| 1 => match self.stat_type { |
| StatsType::Population => Some(0.0), |
| StatsType::Sample => None, |
| }, |
| _ => Some(m2 / count as f64), |
| })) |
| } |
| |
| fn size(&self) -> usize { |
| size_of_val(self) + self.distinct_values.size() |
| } |
| |
| fn state(&mut self) -> Result<Vec<ScalarValue>> { |
| self.distinct_values.state() |
| } |
| |
| fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { |
| self.distinct_values.merge_batch(states) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use datafusion_expr::EmitTo; |
| |
| use super::*; |
| |
| #[test] |
| fn update_batch_ignores_nulls() -> Result<()> { |
| // An array with nulls must accumulate the same values as a dense |
| // array of its non-null values. |
| let dense: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0])); |
| let sparse: ArrayRef = Arc::new(Float64Array::from(vec![ |
| Some(1.0), |
| None, |
| Some(2.0), |
| Some(3.0), |
| None, |
| Some(4.0), |
| ])); |
| |
| let mut dense_acc = VarianceAccumulator::try_new(StatsType::Sample)?; |
| dense_acc.update_batch(std::slice::from_ref(&dense))?; |
| let mut sparse_acc = VarianceAccumulator::try_new(StatsType::Sample)?; |
| sparse_acc.update_batch(std::slice::from_ref(&sparse))?; |
| |
| // Sample variance of {1, 2, 3, 4} is 5/3 (all steps are exact in f64). |
| assert_eq!(dense_acc.evaluate()?, ScalarValue::Float64(Some(5.0 / 3.0))); |
| assert_eq!(dense_acc.evaluate()?, sparse_acc.evaluate()?); |
| Ok(()) |
| } |
| |
| #[test] |
| fn retract_batch_ignores_nulls() -> Result<()> { |
| let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0])); |
| let dense_retract: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0])); |
| let sparse_retract: ArrayRef = |
| Arc::new(Float64Array::from(vec![Some(1.0), None, Some(2.0)])); |
| |
| let mut dense_acc = VarianceAccumulator::try_new(StatsType::Sample)?; |
| dense_acc.update_batch(std::slice::from_ref(&values))?; |
| dense_acc.retract_batch(std::slice::from_ref(&dense_retract))?; |
| let mut sparse_acc = VarianceAccumulator::try_new(StatsType::Sample)?; |
| sparse_acc.update_batch(std::slice::from_ref(&values))?; |
| sparse_acc.retract_batch(std::slice::from_ref(&sparse_retract))?; |
| |
| // Sample variance of the remaining {3, 4} is 0.5 (all steps are exact |
| // in f64). |
| assert_eq!(dense_acc.evaluate()?, ScalarValue::Float64(Some(0.5))); |
| assert_eq!(dense_acc.evaluate()?, sparse_acc.evaluate()?); |
| Ok(()) |
| } |
| |
| #[test] |
| fn retract_batch_resets_when_underflowing() -> Result<()> { |
| // Retracting more values than were accumulated resets to the empty |
| // state, with or without nulls in the retracted batch. |
| let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0])); |
| let dense_retract: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])); |
| let sparse_retract: ArrayRef = Arc::new(Float64Array::from(vec![ |
| Some(1.0), |
| None, |
| Some(2.0), |
| Some(3.0), |
| ])); |
| |
| for retract in [&dense_retract, &sparse_retract] { |
| let mut acc = VarianceAccumulator::try_new(StatsType::Sample)?; |
| acc.update_batch(std::slice::from_ref(&values))?; |
| acc.retract_batch(std::slice::from_ref(retract))?; |
| assert_eq!(acc.get_count(), 0); |
| assert_eq!(acc.evaluate()?, ScalarValue::Float64(None)); |
| } |
| Ok(()) |
| } |
| |
| #[test] |
| fn test_groups_accumulator_merge_empty_states() -> Result<()> { |
| let state_1 = vec![ |
| Arc::new(UInt64Array::from(vec![0])) as ArrayRef, |
| Arc::new(Float64Array::from(vec![0.0])), |
| Arc::new(Float64Array::from(vec![0.0])), |
| ]; |
| let state_2 = vec![ |
| Arc::new(UInt64Array::from(vec![2])) as ArrayRef, |
| Arc::new(Float64Array::from(vec![1.0])), |
| Arc::new(Float64Array::from(vec![1.0])), |
| ]; |
| let mut acc = VarianceGroupsAccumulator::new(StatsType::Sample); |
| acc.merge_batch(&state_1, &[0], 1)?; |
| acc.merge_batch(&state_2, &[0], 1)?; |
| let result = acc.evaluate(EmitTo::All)?; |
| let result = result.as_any().downcast_ref::<Float64Array>().unwrap(); |
| assert_eq!(result.len(), 1); |
| assert_eq!(result.value(0), 1.0); |
| Ok(()) |
| } |
| |
| #[test] |
| fn convert_to_state_roundtrips_through_merge() -> Result<()> { |
| let values = Arc::new(Float64Array::from(vec![ |
| Some(1.0), |
| Some(2.0), |
| None, |
| Some(4.0), |
| Some(8.0), |
| Some(16.0), |
| Some(32.0), |
| ])) as ArrayRef; |
| let filter = BooleanArray::from(vec![ |
| Some(true), |
| Some(false), |
| Some(true), |
| None, |
| Some(true), |
| Some(true), |
| Some(true), |
| ]); |
| let group_indices = vec![0, 1, 0, 1, 0, 0, 0]; |
| |
| let mut direct = VarianceGroupsAccumulator::new(StatsType::Sample); |
| direct.update_batch( |
| std::slice::from_ref(&values), |
| &group_indices, |
| Some(&filter), |
| 2, |
| )?; |
| let direct = direct.evaluate(EmitTo::All)?; |
| |
| let converter = VarianceGroupsAccumulator::new(StatsType::Sample); |
| let state = |
| converter.convert_to_state(std::slice::from_ref(&values), Some(&filter))?; |
| let mut merged = VarianceGroupsAccumulator::new(StatsType::Sample); |
| merged.merge_batch(&state, &group_indices, 2)?; |
| let merged = merged.evaluate(EmitTo::All)?; |
| |
| let direct = direct.as_any().downcast_ref::<Float64Array>().unwrap(); |
| let merged = merged.as_any().downcast_ref::<Float64Array>().unwrap(); |
| assert_eq!(direct.len(), merged.len()); |
| for row in 0..direct.len() { |
| assert_eq!(direct.is_null(row), merged.is_null(row)); |
| if direct.is_valid(row) { |
| assert!((direct.value(row) - merged.value(row)).abs() < 1e-12); |
| } |
| } |
| Ok(()) |
| } |
| |
| #[test] |
| fn convert_to_state_preserves_empty_and_filtered_rows() -> Result<()> { |
| let converter = VarianceGroupsAccumulator::new(StatsType::Sample); |
| let empty_values = |
| Arc::new(Float64Array::from(Vec::<Option<f64>>::new())) as ArrayRef; |
| let state = |
| converter.convert_to_state(std::slice::from_ref(&empty_values), None)?; |
| for state_array in &state { |
| assert_eq!(state_array.len(), 0); |
| assert_eq!(state_array.null_count(), 0); |
| } |
| |
| let values = |
| Arc::new(Float64Array::from(vec![Some(1.0), Some(2.0), None])) as ArrayRef; |
| let filter = BooleanArray::from(vec![Some(false), None, Some(false)]); |
| let group_indices = vec![0, 1, 0]; |
| let state = |
| converter.convert_to_state(std::slice::from_ref(&values), Some(&filter))?; |
| for state_array in &state { |
| assert_eq!(state_array.len(), values.len()); |
| assert_eq!(state_array.null_count(), 0); |
| } |
| |
| let counts = state[0].as_any().downcast_ref::<UInt64Array>().unwrap(); |
| assert_eq!(counts, &UInt64Array::from(vec![0, 0, 0])); |
| |
| let mut merged = VarianceGroupsAccumulator::new(StatsType::Sample); |
| merged.merge_batch(&state, &group_indices, 2)?; |
| let result = merged.evaluate(EmitTo::All)?; |
| let result = result.as_any().downcast_ref::<Float64Array>().unwrap(); |
| assert_eq!(result.len(), 2); |
| assert_eq!(result.null_count(), 2); |
| Ok(()) |
| } |
| } |
