datafusion-functions-aggregate 55.0.0

// 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(())
    }
}