datafusion_functions_aggregate/
variance.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! [`VarianceSample`]: variance sample aggregations.
19//! [`VariancePopulation`]: variance population aggregations.
20
21use arrow::datatypes::{FieldRef, Float64Type};
22use arrow::{
23    array::{Array, ArrayRef, BooleanArray, Float64Array, UInt64Array},
24    buffer::NullBuffer,
25    datatypes::{DataType, Field},
26};
27use datafusion_common::cast::{as_float64_array, as_uint64_array};
28use datafusion_common::{Result, ScalarValue};
29use datafusion_expr::{
30    Accumulator, AggregateUDFImpl, Documentation, GroupsAccumulator, Signature,
31    Volatility,
32    function::{AccumulatorArgs, StateFieldsArgs},
33    utils::format_state_name,
34};
35use datafusion_functions_aggregate_common::utils::GenericDistinctBuffer;
36use datafusion_functions_aggregate_common::{
37    aggregate::groups_accumulator::accumulate::accumulate, stats::StatsType,
38};
39use datafusion_macros::user_doc;
40use std::mem::{size_of, size_of_val};
41use std::{fmt::Debug, sync::Arc};
42
43make_udaf_expr_and_func!(
44    VarianceSample,
45    var_sample,
46    expression,
47    "Computes the sample variance.",
48    var_samp_udaf
49);
50
51make_udaf_expr_and_func!(
52    VariancePopulation,
53    var_pop,
54    expression,
55    "Computes the population variance.",
56    var_pop_udaf
57);
58
59#[user_doc(
60    doc_section(label = "General Functions"),
61    description = "Returns the statistical sample variance of a set of numbers.",
62    syntax_example = "var(expression)",
63    standard_argument(name = "expression", prefix = "Numeric")
64)]
65#[derive(PartialEq, Eq, Hash, Debug)]
66pub struct VarianceSample {
67    signature: Signature,
68    aliases: Vec<String>,
69}
70
71impl Default for VarianceSample {
72    fn default() -> Self {
73        Self::new()
74    }
75}
76
77impl VarianceSample {
78    pub fn new() -> Self {
79        Self {
80            aliases: vec![String::from("var_sample"), String::from("var_samp")],
81            signature: Signature::exact(vec![DataType::Float64], Volatility::Immutable),
82        }
83    }
84}
85
86impl AggregateUDFImpl for VarianceSample {
87    fn name(&self) -> &str {
88        "var"
89    }
90
91    fn signature(&self) -> &Signature {
92        &self.signature
93    }
94
95    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
96        Ok(DataType::Float64)
97    }
98
99    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
100        let name = args.name;
101        match args.is_distinct {
102            false => Ok(vec![
103                Field::new(format_state_name(name, "count"), DataType::UInt64, true),
104                Field::new(format_state_name(name, "mean"), DataType::Float64, true),
105                Field::new(format_state_name(name, "m2"), DataType::Float64, true),
106            ]
107            .into_iter()
108            .map(Arc::new)
109            .collect()),
110            true => {
111                let field = Field::new_list_field(DataType::Float64, true);
112                let state_name = "distinct_var";
113                Ok(vec![
114                    Field::new(
115                        format_state_name(name, state_name),
116                        DataType::List(Arc::new(field)),
117                        true,
118                    )
119                    .into(),
120                ])
121            }
122        }
123    }
124
125    fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
126        if acc_args.is_distinct {
127            return Ok(Box::new(DistinctVarianceAccumulator::new(
128                StatsType::Sample,
129            )));
130        }
131
132        Ok(Box::new(VarianceAccumulator::try_new(StatsType::Sample)?))
133    }
134
135    fn aliases(&self) -> &[String] {
136        &self.aliases
137    }
138
139    fn groups_accumulator_supported(&self, acc_args: AccumulatorArgs) -> bool {
140        !acc_args.is_distinct
141    }
142
143    fn create_groups_accumulator(
144        &self,
145        _args: AccumulatorArgs,
146    ) -> Result<Box<dyn GroupsAccumulator>> {
147        Ok(Box::new(VarianceGroupsAccumulator::new(StatsType::Sample)))
148    }
149
150    fn documentation(&self) -> Option<&Documentation> {
151        self.doc()
152    }
153}
154
155#[user_doc(
156    doc_section(label = "General Functions"),
157    description = "Returns the statistical population variance of a set of numbers.",
158    syntax_example = "var_pop(expression)",
159    standard_argument(name = "expression", prefix = "Numeric")
160)]
161#[derive(PartialEq, Eq, Hash, Debug)]
162pub struct VariancePopulation {
163    signature: Signature,
164    aliases: Vec<String>,
165}
166
167impl Default for VariancePopulation {
168    fn default() -> Self {
169        Self::new()
170    }
171}
172
173impl VariancePopulation {
174    pub fn new() -> Self {
175        Self {
176            aliases: vec![String::from("var_population")],
177            signature: Signature::exact(vec![DataType::Float64], Volatility::Immutable),
178        }
179    }
180}
181
182impl AggregateUDFImpl for VariancePopulation {
183    fn name(&self) -> &str {
184        "var_pop"
185    }
186
187    fn signature(&self) -> &Signature {
188        &self.signature
189    }
190
191    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
192        Ok(DataType::Float64)
193    }
194
195    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
196        match args.is_distinct {
197            false => {
198                let name = args.name;
199                Ok(vec![
200                    Field::new(format_state_name(name, "count"), DataType::UInt64, true),
201                    Field::new(format_state_name(name, "mean"), DataType::Float64, true),
202                    Field::new(format_state_name(name, "m2"), DataType::Float64, true),
203                ]
204                .into_iter()
205                .map(Arc::new)
206                .collect())
207            }
208            true => {
209                let field = Field::new_list_field(DataType::Float64, true);
210                let state_name = "distinct_var";
211                Ok(vec![
212                    Field::new(
213                        format_state_name(args.name, state_name),
214                        DataType::List(Arc::new(field)),
215                        true,
216                    )
217                    .into(),
218                ])
219            }
220        }
221    }
222
223    fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
224        if acc_args.is_distinct {
225            return Ok(Box::new(DistinctVarianceAccumulator::new(
226                StatsType::Population,
227            )));
228        }
229
230        Ok(Box::new(VarianceAccumulator::try_new(
231            StatsType::Population,
232        )?))
233    }
234
235    fn aliases(&self) -> &[String] {
236        &self.aliases
237    }
238
239    fn groups_accumulator_supported(&self, acc_args: AccumulatorArgs) -> bool {
240        !acc_args.is_distinct
241    }
242
243    fn create_groups_accumulator(
244        &self,
245        _args: AccumulatorArgs,
246    ) -> Result<Box<dyn GroupsAccumulator>> {
247        Ok(Box::new(VarianceGroupsAccumulator::new(
248            StatsType::Population,
249        )))
250    }
251
252    fn documentation(&self) -> Option<&Documentation> {
253        self.doc()
254    }
255}
256
257/// An accumulator to compute variance
258/// The algorithm used is an online implementation and numerically stable. It is based on this paper:
259/// Welford, B. P. (1962). "Note on a method for calculating corrected sums of squares and products".
260/// Technometrics. 4 (3): 419–420. doi:10.2307/1266577. JSTOR 1266577.
261///
262/// The algorithm has been analyzed here:
263/// Ling, Robert F. (1974). "Comparison of Several Algorithms for Computing Sample Means and Variances".
264/// Journal of the American Statistical Association. 69 (348): 859–866. doi:10.2307/2286154. JSTOR 2286154.
265
266#[derive(Debug)]
267pub struct VarianceAccumulator {
268    m2: f64,
269    mean: f64,
270    count: u64,
271    stats_type: StatsType,
272}
273
274impl VarianceAccumulator {
275    /// Creates a new `VarianceAccumulator`
276    pub fn try_new(s_type: StatsType) -> Result<Self> {
277        Ok(Self {
278            m2: 0_f64,
279            mean: 0_f64,
280            count: 0_u64,
281            stats_type: s_type,
282        })
283    }
284
285    pub fn get_count(&self) -> u64 {
286        self.count
287    }
288
289    pub fn get_mean(&self) -> f64 {
290        self.mean
291    }
292
293    pub fn get_m2(&self) -> f64 {
294        self.m2
295    }
296}
297
298#[inline]
299fn merge(
300    count: u64,
301    mean: f64,
302    m2: f64,
303    count2: u64,
304    mean2: f64,
305    m22: f64,
306) -> (u64, f64, f64) {
307    debug_assert!(count != 0 || count2 != 0, "Cannot merge two empty states");
308    let new_count = count + count2;
309    let new_mean =
310        mean * count as f64 / new_count as f64 + mean2 * count2 as f64 / new_count as f64;
311    let delta = mean - mean2;
312    let new_m2 =
313        m2 + m22 + delta * delta * count as f64 * count2 as f64 / new_count as f64;
314
315    (new_count, new_mean, new_m2)
316}
317
318#[inline]
319fn update(count: u64, mean: f64, m2: f64, value: f64) -> (u64, f64, f64) {
320    let new_count = count + 1;
321    let delta1 = value - mean;
322    let new_mean = delta1 / new_count as f64 + mean;
323    let delta2 = value - new_mean;
324    let new_m2 = m2 + delta1 * delta2;
325
326    (new_count, new_mean, new_m2)
327}
328
329/// Inverse of [`update`]: removes a previously accumulated value. Retracting
330/// from a state with one or zero values resets the state to empty.
331#[inline]
332fn retract(count: u64, mean: f64, m2: f64, value: f64) -> (u64, f64, f64) {
333    if count <= 1 {
334        return (0, 0.0, 0.0);
335    }
336
337    let new_count = count - 1;
338    let delta1 = mean - value;
339    let new_mean = delta1 / new_count as f64 + mean;
340    let delta2 = new_mean - value;
341    let new_m2 = m2 - delta1 * delta2;
342
343    (new_count, new_mean, new_m2)
344}
345
346impl Accumulator for VarianceAccumulator {
347    fn state(&mut self) -> Result<Vec<ScalarValue>> {
348        Ok(vec![
349            ScalarValue::from(self.count),
350            ScalarValue::from(self.mean),
351            ScalarValue::from(self.m2),
352        ])
353    }
354
355    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
356        let arr = as_float64_array(&values[0])?;
357        for value in arr.iter().flatten() {
358            (self.count, self.mean, self.m2) =
359                update(self.count, self.mean, self.m2, value)
360        }
361
362        Ok(())
363    }
364
365    fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
366        let arr = as_float64_array(&values[0])?;
367        for value in arr.iter().flatten() {
368            (self.count, self.mean, self.m2) =
369                retract(self.count, self.mean, self.m2, value)
370        }
371
372        Ok(())
373    }
374
375    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
376        let counts = as_uint64_array(&states[0])?;
377        let means = as_float64_array(&states[1])?;
378        let m2s = as_float64_array(&states[2])?;
379
380        for i in 0..counts.len() {
381            let c = counts.value(i);
382            if c == 0_u64 {
383                continue;
384            }
385            (self.count, self.mean, self.m2) = merge(
386                self.count,
387                self.mean,
388                self.m2,
389                c,
390                means.value(i),
391                m2s.value(i),
392            )
393        }
394        Ok(())
395    }
396
397    fn evaluate(&mut self) -> Result<ScalarValue> {
398        let count = match self.stats_type {
399            StatsType::Population => self.count,
400            StatsType::Sample => {
401                if self.count > 0 {
402                    self.count - 1
403                } else {
404                    self.count
405                }
406            }
407        };
408
409        Ok(ScalarValue::Float64(match self.count {
410            0 => None,
411            1 => {
412                if let StatsType::Population = self.stats_type {
413                    Some(0.0)
414                } else {
415                    None
416                }
417            }
418            _ => Some(self.m2 / count as f64),
419        }))
420    }
421
422    fn size(&self) -> usize {
423        size_of_val(self)
424    }
425
426    fn supports_retract_batch(&self) -> bool {
427        true
428    }
429}
430
431#[derive(Debug)]
432pub struct VarianceGroupsAccumulator {
433    m2s: Vec<f64>,
434    means: Vec<f64>,
435    counts: Vec<u64>,
436    stats_type: StatsType,
437}
438
439impl VarianceGroupsAccumulator {
440    pub fn new(s_type: StatsType) -> Self {
441        Self {
442            m2s: Vec::new(),
443            means: Vec::new(),
444            counts: Vec::new(),
445            stats_type: s_type,
446        }
447    }
448
449    fn resize(&mut self, total_num_groups: usize) {
450        self.m2s.resize(total_num_groups, 0.0);
451        self.means.resize(total_num_groups, 0.0);
452        self.counts.resize(total_num_groups, 0);
453    }
454
455    fn merge<F>(
456        group_indices: &[usize],
457        counts: &UInt64Array,
458        means: &Float64Array,
459        m2s: &Float64Array,
460        _opt_filter: Option<&BooleanArray>,
461        mut value_fn: F,
462    ) where
463        F: FnMut(usize, u64, f64, f64) + Send,
464    {
465        assert_eq!(counts.null_count(), 0);
466        assert_eq!(means.null_count(), 0);
467        assert_eq!(m2s.null_count(), 0);
468
469        group_indices
470            .iter()
471            .zip(counts.values().iter())
472            .zip(means.values().iter())
473            .zip(m2s.values().iter())
474            .for_each(|(((&group_index, &count), &mean), &m2)| {
475                value_fn(group_index, count, mean, m2);
476            });
477    }
478
479    pub fn variance(
480        &mut self,
481        emit_to: datafusion_expr::EmitTo,
482    ) -> (Vec<f64>, NullBuffer) {
483        let mut counts = emit_to.take_needed(&mut self.counts);
484        // means are only needed for updating m2s and are not needed for the final result.
485        // But we still need to take them to ensure the internal state is consistent.
486        let _ = emit_to.take_needed(&mut self.means);
487        let m2s = emit_to.take_needed(&mut self.m2s);
488
489        if let StatsType::Sample = self.stats_type {
490            counts.iter_mut().for_each(|count| {
491                *count = count.saturating_sub(1);
492            });
493        }
494        let nulls = NullBuffer::from_iter(counts.iter().map(|&count| count != 0));
495        let variance = m2s
496            .iter()
497            .zip(counts)
498            .map(|(m2, count)| m2 / count as f64)
499            .collect();
500        (variance, nulls)
501    }
502}
503
504impl GroupsAccumulator for VarianceGroupsAccumulator {
505    fn update_batch(
506        &mut self,
507        values: &[ArrayRef],
508        group_indices: &[usize],
509        opt_filter: Option<&BooleanArray>,
510        total_num_groups: usize,
511    ) -> Result<()> {
512        assert_eq!(values.len(), 1, "single argument to update_batch");
513        let values = as_float64_array(&values[0])?;
514
515        self.resize(total_num_groups);
516        accumulate(group_indices, values, opt_filter, |group_index, value| {
517            let (new_count, new_mean, new_m2) = update(
518                self.counts[group_index],
519                self.means[group_index],
520                self.m2s[group_index],
521                value,
522            );
523            self.counts[group_index] = new_count;
524            self.means[group_index] = new_mean;
525            self.m2s[group_index] = new_m2;
526        });
527        Ok(())
528    }
529
530    fn merge_batch(
531        &mut self,
532        values: &[ArrayRef],
533        group_indices: &[usize],
534        total_num_groups: usize,
535    ) -> Result<()> {
536        assert_eq!(values.len(), 3, "two arguments to merge_batch");
537        // first batch is counts, second is partial means, third is partial m2s
538        let partial_counts = as_uint64_array(&values[0])?;
539        let partial_means = as_float64_array(&values[1])?;
540        let partial_m2s = as_float64_array(&values[2])?;
541
542        self.resize(total_num_groups);
543        Self::merge(
544            group_indices,
545            partial_counts,
546            partial_means,
547            partial_m2s,
548            None,
549            |group_index, partial_count, partial_mean, partial_m2| {
550                if partial_count == 0 {
551                    return;
552                }
553                let (new_count, new_mean, new_m2) = merge(
554                    self.counts[group_index],
555                    self.means[group_index],
556                    self.m2s[group_index],
557                    partial_count,
558                    partial_mean,
559                    partial_m2,
560                );
561                self.counts[group_index] = new_count;
562                self.means[group_index] = new_mean;
563                self.m2s[group_index] = new_m2;
564            },
565        );
566        Ok(())
567    }
568
569    fn evaluate(&mut self, emit_to: datafusion_expr::EmitTo) -> Result<ArrayRef> {
570        let (variances, nulls) = self.variance(emit_to);
571        Ok(Arc::new(Float64Array::new(variances.into(), Some(nulls))))
572    }
573
574    fn state(&mut self, emit_to: datafusion_expr::EmitTo) -> Result<Vec<ArrayRef>> {
575        let counts = emit_to.take_needed(&mut self.counts);
576        let means = emit_to.take_needed(&mut self.means);
577        let m2s = emit_to.take_needed(&mut self.m2s);
578
579        Ok(vec![
580            Arc::new(UInt64Array::new(counts.into(), None)),
581            Arc::new(Float64Array::new(means.into(), None)),
582            Arc::new(Float64Array::new(m2s.into(), None)),
583        ])
584    }
585
586    fn convert_to_state(
587        &self,
588        values: &[ArrayRef],
589        opt_filter: Option<&BooleanArray>,
590    ) -> Result<Vec<ArrayRef>> {
591        assert_eq!(values.len(), 1, "single argument to convert_to_state");
592        let values = as_float64_array(&values[0])?;
593
594        let len = values.len();
595        let mut counts = Vec::with_capacity(len);
596        let mut means = Vec::with_capacity(len);
597        let mut m2s = Vec::with_capacity(len);
598
599        for row in 0..len {
600            if values.is_valid(row)
601                && opt_filter
602                    .is_none_or(|filter| filter.is_valid(row) && filter.value(row))
603            {
604                counts.push(1);
605                means.push(values.value(row));
606            } else {
607                counts.push(0);
608                means.push(0.0);
609            }
610            m2s.push(0.0);
611        }
612
613        Ok(vec![
614            Arc::new(UInt64Array::new(counts.into(), None)),
615            Arc::new(Float64Array::new(means.into(), None)),
616            Arc::new(Float64Array::new(m2s.into(), None)),
617        ])
618    }
619    fn size(&self) -> usize {
620        self.m2s.capacity() * size_of::<f64>()
621            + self.means.capacity() * size_of::<f64>()
622            + self.counts.capacity() * size_of::<u64>()
623    }
624}
625
626#[derive(Debug)]
627pub struct DistinctVarianceAccumulator {
628    distinct_values: GenericDistinctBuffer<Float64Type>,
629    stat_type: StatsType,
630}
631
632impl DistinctVarianceAccumulator {
633    pub fn new(stat_type: StatsType) -> Self {
634        Self {
635            distinct_values: GenericDistinctBuffer::<Float64Type>::new(DataType::Float64),
636            stat_type,
637        }
638    }
639}
640
641impl Accumulator for DistinctVarianceAccumulator {
642    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
643        self.distinct_values.update_batch(values)
644    }
645
646    fn evaluate(&mut self) -> Result<ScalarValue> {
647        let values = self
648            .distinct_values
649            .values
650            .iter()
651            .map(|v| v.0)
652            .collect::<Vec<_>>();
653
654        let count = match self.stat_type {
655            StatsType::Sample => {
656                if !values.is_empty() {
657                    values.len() - 1
658                } else {
659                    0
660                }
661            }
662            StatsType::Population => values.len(),
663        };
664
665        let mean = values.iter().sum::<f64>() / values.len() as f64;
666        let m2 = values.iter().map(|x| (x - mean) * (x - mean)).sum::<f64>();
667
668        Ok(ScalarValue::Float64(match values.len() {
669            0 => None,
670            1 => match self.stat_type {
671                StatsType::Population => Some(0.0),
672                StatsType::Sample => None,
673            },
674            _ => Some(m2 / count as f64),
675        }))
676    }
677
678    fn size(&self) -> usize {
679        size_of_val(self) + self.distinct_values.size()
680    }
681
682    fn state(&mut self) -> Result<Vec<ScalarValue>> {
683        self.distinct_values.state()
684    }
685
686    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
687        self.distinct_values.merge_batch(states)
688    }
689}
690
691#[cfg(test)]
692mod tests {
693    use datafusion_expr::EmitTo;
694
695    use super::*;
696
697    #[test]
698    fn update_batch_ignores_nulls() -> Result<()> {
699        // An array with nulls must accumulate the same values as a dense
700        // array of its non-null values.
701        let dense: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0]));
702        let sparse: ArrayRef = Arc::new(Float64Array::from(vec![
703            Some(1.0),
704            None,
705            Some(2.0),
706            Some(3.0),
707            None,
708            Some(4.0),
709        ]));
710
711        let mut dense_acc = VarianceAccumulator::try_new(StatsType::Sample)?;
712        dense_acc.update_batch(std::slice::from_ref(&dense))?;
713        let mut sparse_acc = VarianceAccumulator::try_new(StatsType::Sample)?;
714        sparse_acc.update_batch(std::slice::from_ref(&sparse))?;
715
716        // Sample variance of {1, 2, 3, 4} is 5/3 (all steps are exact in f64).
717        assert_eq!(dense_acc.evaluate()?, ScalarValue::Float64(Some(5.0 / 3.0)));
718        assert_eq!(dense_acc.evaluate()?, sparse_acc.evaluate()?);
719        Ok(())
720    }
721
722    #[test]
723    fn retract_batch_ignores_nulls() -> Result<()> {
724        let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0]));
725        let dense_retract: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0]));
726        let sparse_retract: ArrayRef =
727            Arc::new(Float64Array::from(vec![Some(1.0), None, Some(2.0)]));
728
729        let mut dense_acc = VarianceAccumulator::try_new(StatsType::Sample)?;
730        dense_acc.update_batch(std::slice::from_ref(&values))?;
731        dense_acc.retract_batch(std::slice::from_ref(&dense_retract))?;
732        let mut sparse_acc = VarianceAccumulator::try_new(StatsType::Sample)?;
733        sparse_acc.update_batch(std::slice::from_ref(&values))?;
734        sparse_acc.retract_batch(std::slice::from_ref(&sparse_retract))?;
735
736        // Sample variance of the remaining {3, 4} is 0.5 (all steps are exact
737        // in f64).
738        assert_eq!(dense_acc.evaluate()?, ScalarValue::Float64(Some(0.5)));
739        assert_eq!(dense_acc.evaluate()?, sparse_acc.evaluate()?);
740        Ok(())
741    }
742
743    #[test]
744    fn retract_batch_resets_when_underflowing() -> Result<()> {
745        // Retracting more values than were accumulated resets to the empty
746        // state, with or without nulls in the retracted batch.
747        let values: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0]));
748        let dense_retract: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0]));
749        let sparse_retract: ArrayRef = Arc::new(Float64Array::from(vec![
750            Some(1.0),
751            None,
752            Some(2.0),
753            Some(3.0),
754        ]));
755
756        for retract in [&dense_retract, &sparse_retract] {
757            let mut acc = VarianceAccumulator::try_new(StatsType::Sample)?;
758            acc.update_batch(std::slice::from_ref(&values))?;
759            acc.retract_batch(std::slice::from_ref(retract))?;
760            assert_eq!(acc.get_count(), 0);
761            assert_eq!(acc.evaluate()?, ScalarValue::Float64(None));
762        }
763        Ok(())
764    }
765
766    #[test]
767    fn test_groups_accumulator_merge_empty_states() -> Result<()> {
768        let state_1 = vec![
769            Arc::new(UInt64Array::from(vec![0])) as ArrayRef,
770            Arc::new(Float64Array::from(vec![0.0])),
771            Arc::new(Float64Array::from(vec![0.0])),
772        ];
773        let state_2 = vec![
774            Arc::new(UInt64Array::from(vec![2])) as ArrayRef,
775            Arc::new(Float64Array::from(vec![1.0])),
776            Arc::new(Float64Array::from(vec![1.0])),
777        ];
778        let mut acc = VarianceGroupsAccumulator::new(StatsType::Sample);
779        acc.merge_batch(&state_1, &[0], 1)?;
780        acc.merge_batch(&state_2, &[0], 1)?;
781        let result = acc.evaluate(EmitTo::All)?;
782        let result = result.as_any().downcast_ref::<Float64Array>().unwrap();
783        assert_eq!(result.len(), 1);
784        assert_eq!(result.value(0), 1.0);
785        Ok(())
786    }
787
788    #[test]
789    fn convert_to_state_roundtrips_through_merge() -> Result<()> {
790        let values = Arc::new(Float64Array::from(vec![
791            Some(1.0),
792            Some(2.0),
793            None,
794            Some(4.0),
795            Some(8.0),
796            Some(16.0),
797            Some(32.0),
798        ])) as ArrayRef;
799        let filter = BooleanArray::from(vec![
800            Some(true),
801            Some(false),
802            Some(true),
803            None,
804            Some(true),
805            Some(true),
806            Some(true),
807        ]);
808        let group_indices = vec![0, 1, 0, 1, 0, 0, 0];
809
810        let mut direct = VarianceGroupsAccumulator::new(StatsType::Sample);
811        direct.update_batch(
812            std::slice::from_ref(&values),
813            &group_indices,
814            Some(&filter),
815            2,
816        )?;
817        let direct = direct.evaluate(EmitTo::All)?;
818
819        let converter = VarianceGroupsAccumulator::new(StatsType::Sample);
820        let state =
821            converter.convert_to_state(std::slice::from_ref(&values), Some(&filter))?;
822        let mut merged = VarianceGroupsAccumulator::new(StatsType::Sample);
823        merged.merge_batch(&state, &group_indices, 2)?;
824        let merged = merged.evaluate(EmitTo::All)?;
825
826        let direct = direct.as_any().downcast_ref::<Float64Array>().unwrap();
827        let merged = merged.as_any().downcast_ref::<Float64Array>().unwrap();
828        assert_eq!(direct.len(), merged.len());
829        for row in 0..direct.len() {
830            assert_eq!(direct.is_null(row), merged.is_null(row));
831            if direct.is_valid(row) {
832                assert!((direct.value(row) - merged.value(row)).abs() < 1e-12);
833            }
834        }
835        Ok(())
836    }
837
838    #[test]
839    fn convert_to_state_preserves_empty_and_filtered_rows() -> Result<()> {
840        let converter = VarianceGroupsAccumulator::new(StatsType::Sample);
841        let empty_values =
842            Arc::new(Float64Array::from(Vec::<Option<f64>>::new())) as ArrayRef;
843        let state =
844            converter.convert_to_state(std::slice::from_ref(&empty_values), None)?;
845        for state_array in &state {
846            assert_eq!(state_array.len(), 0);
847            assert_eq!(state_array.null_count(), 0);
848        }
849
850        let values =
851            Arc::new(Float64Array::from(vec![Some(1.0), Some(2.0), None])) as ArrayRef;
852        let filter = BooleanArray::from(vec![Some(false), None, Some(false)]);
853        let group_indices = vec![0, 1, 0];
854        let state =
855            converter.convert_to_state(std::slice::from_ref(&values), Some(&filter))?;
856        for state_array in &state {
857            assert_eq!(state_array.len(), values.len());
858            assert_eq!(state_array.null_count(), 0);
859        }
860
861        let counts = state[0].as_any().downcast_ref::<UInt64Array>().unwrap();
862        assert_eq!(counts, &UInt64Array::from(vec![0, 0, 0]));
863
864        let mut merged = VarianceGroupsAccumulator::new(StatsType::Sample);
865        merged.merge_batch(&state, &group_indices, 2)?;
866        let result = merged.evaluate(EmitTo::All)?;
867        let result = result.as_any().downcast_ref::<Float64Array>().unwrap();
868        assert_eq!(result.len(), 2);
869        assert_eq!(result.null_count(), 2);
870        Ok(())
871    }
872}
datafusion_functions_aggregate/variance.rs

datafusion_functions_aggregate/
variance.rs
