linkrs-query 0.1.0

use std::collections::HashSet;

use super::comparison::compare_values;
use linkrs_core::types::expr::Expression;
use linkrs_core::types::operators::AggregateFunction;
use linkrs_core::value::NullType;
use linkrs_core::Value;

/// Extract a numeric parameter (e.g. the percentile fraction) from the
/// aggregate's `args` list. Parameters beyond the field expression live at
/// index 1 (index 0 is the aggregated field itself).
fn aggregate_param_f64(args: &[Expression], idx: usize) -> Option<f64> {
    match args.get(idx) {
        Some(Expression::Literal(Value::Int(v))) => Some(*v as f64),
        Some(Expression::Literal(Value::BigInt(v))) => Some(*v as f64),
        Some(Expression::Literal(Value::Float(v))) => Some(*v as f64),
        Some(Expression::Literal(Value::Double(v))) => Some(*v),
        _ => None,
    }
}

/// Extract a string parameter (e.g. the GROUP_CONCAT separator) from the
/// aggregate's `args` list (index 1).
fn aggregate_param_string(args: &[Expression], idx: usize) -> Option<String> {
    match args.get(idx) {
        Some(Expression::Literal(Value::String(s))) => Some(s.to_string()),
        _ => None,
    }
}

/// Typed accumulator state for aggregate results.
///
/// Every [`AggregateFunction`] variant has an accumulator so that the
/// streaming Aggregate, PartialAggregate and spill paths share a single
/// implementation; there is no row-based fallback.
///
/// The Welford variants (`Std`/`StddevPop`/`StddevSamp`/`Variance`) keep
/// `(n, mean, m2)` so that partial states can be merged exactly.
#[derive(Debug, Clone)]
pub enum AggregateAccumulator {
    Count(u64),
    Sum(f64),
    Min(Option<Value>),
    Max(Option<Value>),
    Avg {
        sum: f64,
        count: u64,
    },
    Collect(Vec<Value>),
    CollectSet(HashSet<Value>),
    Distinct(HashSet<Value>),
    Percentile {
        values: Vec<f64>,
        percentile: f64,
    },
    PercentileCont {
        values: Vec<f64>,
        percentile: f64,
    },
    Median(Vec<f64>),
    Mode(Vec<Value>),
    Std {
        n: u64,
        mean: f64,
        m2: f64,
    },
    StddevPop {
        n: u64,
        mean: f64,
        m2: f64,
    },
    StddevSamp {
        n: u64,
        mean: f64,
        m2: f64,
    },
    Variance {
        n: u64,
        mean: f64,
        m2: f64,
    },
    Product(Option<f64>),
    BitAnd(Option<i64>),
    BitOr(Option<i64>),
    BoolAnd(Option<bool>),
    BoolOr(Option<bool>),
    GroupConcat {
        parts: Vec<String>,
        separator: String,
    },
    GroupConcatWithOrder {
        parts: Vec<String>,
        separator: String,
    },
    VecSum(Option<Vec<f32>>),
    VecAvg {
        sum: Vec<f32>,
        count: u64,
    },
}

impl AggregateAccumulator {
    pub fn for_function(func: &AggregateFunction, args: &[Expression]) -> Option<Self> {
        match func {
            AggregateFunction::Count => Some(Self::Count(0)),
            AggregateFunction::Sum => Some(Self::Sum(0.0)),
            AggregateFunction::Min => Some(Self::Min(None)),
            AggregateFunction::Max => Some(Self::Max(None)),
            AggregateFunction::Avg => Some(Self::Avg { sum: 0.0, count: 0 }),
            AggregateFunction::Collect => Some(Self::Collect(Vec::new())),
            AggregateFunction::CollectSet => Some(Self::CollectSet(HashSet::new())),
            AggregateFunction::Percentile => Some(Self::Percentile {
                values: Vec::new(),
                percentile: aggregate_param_f64(args, 1).unwrap_or(50.0),
            }),
            AggregateFunction::PercentileCont => Some(Self::PercentileCont {
                values: Vec::new(),
                percentile: aggregate_param_f64(args, 1).unwrap_or(50.0),
            }),
            AggregateFunction::Median => Some(Self::Median(Vec::new())),
            AggregateFunction::Mode => Some(Self::Mode(Vec::new())),
            AggregateFunction::Std => Some(Self::Std {
                n: 0,
                mean: 0.0,
                m2: 0.0,
            }),
            AggregateFunction::StddevPop => Some(Self::StddevPop {
                n: 0,
                mean: 0.0,
                m2: 0.0,
            }),
            AggregateFunction::StddevSamp => Some(Self::StddevSamp {
                n: 0,
                mean: 0.0,
                m2: 0.0,
            }),
            AggregateFunction::Variance => Some(Self::Variance {
                n: 0,
                mean: 0.0,
                m2: 0.0,
            }),
            AggregateFunction::Product => Some(Self::Product(None)),
            AggregateFunction::BitAnd => Some(Self::BitAnd(None)),
            AggregateFunction::BitOr => Some(Self::BitOr(None)),
            AggregateFunction::BoolAnd => Some(Self::BoolAnd(None)),
            AggregateFunction::BoolOr => Some(Self::BoolOr(None)),
            AggregateFunction::GroupConcat => Some(Self::GroupConcat {
                parts: Vec::new(),
                separator: aggregate_param_string(args, 1).unwrap_or_else(|| ",".to_string()),
            }),
            AggregateFunction::GroupConcatWithOrder => Some(Self::GroupConcatWithOrder {
                parts: Vec::new(),
                separator: aggregate_param_string(args, 1).unwrap_or_else(|| ",".to_string()),
            }),
            AggregateFunction::VecSum => Some(Self::VecSum(None)),
            AggregateFunction::VecAvg => Some(Self::VecAvg {
                sum: Vec::new(),
                count: 0,
            }),
        }
    }

    /// Accumulate a single value into this state.
    pub fn accumulate(&mut self, value: &Value) {
        match self {
            Self::Count(count) => {
                if !matches!(value, Value::Null(_)) {
                    *count += 1;
                }
            }
            Self::Sum(sum) => {
                if let Some(n) = numeric_value(value) {
                    *sum += n;
                }
            }
            Self::Min(ref mut current) => {
                if matches!(value, Value::Null(_)) {
                    return;
                }
                match current {
                    None => *current = Some(value.clone()),
                    Some(c) => {
                        if compare_values(value, c).is_lt() {
                            *current = Some(value.clone());
                        }
                    }
                }
            }
            Self::Max(ref mut current) => {
                if matches!(value, Value::Null(_)) {
                    return;
                }
                match current {
                    None => *current = Some(value.clone()),
                    Some(c) => {
                        if compare_values(value, c).is_gt() {
                            *current = Some(value.clone());
                        }
                    }
                }
            }
            Self::Avg { sum, count } => {
                if let Some(n) = numeric_value(value) {
                    *sum += n;
                    *count += 1;
                }
            }
            Self::Collect(values) => {
                if !matches!(value, Value::Null(_)) {
                    values.push(value.clone());
                }
            }
            Self::CollectSet(set) | Self::Distinct(set) => {
                if !matches!(value, Value::Null(_)) {
                    set.insert(value.clone());
                }
            }
            Self::Percentile { values, .. } | Self::PercentileCont { values, .. } => {
                if let Some(n) = numeric_value(value) {
                    values.push(n);
                }
            }
            Self::Median(values) => {
                if let Some(n) = numeric_value(value) {
                    values.push(n);
                }
            }
            Self::Mode(values) => {
                if !matches!(value, Value::Null(_)) {
                    values.push(value.clone());
                }
            }
            Self::Std { n, mean, m2 }
            | Self::StddevPop { n, mean, m2 }
            | Self::StddevSamp { n, mean, m2 }
            | Self::Variance { n, mean, m2 } => {
                if let Some(x) = numeric_value(value) {
                    welford_update(n, mean, m2, x);
                }
            }
            Self::Product(product) => {
                if let Some(n) = numeric_value(value) {
                    match product {
                        None => *product = Some(n),
                        Some(p) => *p *= n,
                    }
                }
            }
            Self::BitAnd(current) => {
                if let Value::BigInt(v) = value {
                    *current = Some(current.unwrap_or(i64::MAX) & v);
                }
            }
            Self::BitOr(current) => {
                if let Value::BigInt(v) = value {
                    *current = Some(current.unwrap_or(0) | v);
                }
            }
            Self::BoolAnd(current) => {
                if let Value::Bool(b) = value {
                    *current = Some(current.unwrap_or(true) && *b);
                }
            }
            Self::BoolOr(current) => {
                if let Value::Bool(b) = value {
                    *current = Some(current.unwrap_or(false) || *b);
                }
            }
            Self::GroupConcat { parts, .. } | Self::GroupConcatWithOrder { parts, .. } => {
                if !matches!(value, Value::Null(_)) {
                    parts.push(format!("{}", value));
                }
            }
            Self::VecSum(sum) => {
                if let Value::Vector(vec) = value {
                    let dense = vec.to_dense();
                    match sum {
                        None => *sum = Some(dense),
                        Some(s) => {
                            if s.len() == dense.len() {
                                for (a, b) in s.iter_mut().zip(dense) {
                                    *a += b;
                                }
                            }
                        }
                    }
                }
            }
            Self::VecAvg { sum, count } => {
                if let Value::Vector(vec) = value {
                    let dense = vec.to_dense();
                    *count += 1;
                    if sum.is_empty() {
                        *sum = dense;
                    } else if sum.len() == dense.len() {
                        for (x, y) in sum.iter_mut().zip(dense) {
                            *x += y;
                        }
                    }
                }
            }
        }
    }

    /// Merge another accumulator of the same kind into this one.
    pub fn merge(&mut self, other: &AggregateAccumulator) {
        match (self, other) {
            (Self::Count(a), Self::Count(b)) => *a += b,
            (Self::Sum(a), Self::Sum(b)) => *a += b,
            (Self::Min(ref mut a), Self::Min(b)) => match (a.as_ref(), b) {
                (None, Some(v)) => *a = Some(v.clone()),
                (Some(_), None) => {}
                (Some(a_val), Some(b_val)) => {
                    if compare_values(b_val, a_val).is_lt() {
                        *a = Some(b_val.clone());
                    }
                }
                (None, None) => {}
            },
            (Self::Max(ref mut a), Self::Max(b)) => match (a.as_ref(), b) {
                (None, Some(v)) => *a = Some(v.clone()),
                (Some(_), None) => {}
                (Some(a_val), Some(b_val)) => {
                    if compare_values(b_val, a_val).is_gt() {
                        *a = Some(b_val.clone());
                    }
                }
                (None, None) => {}
            },
            (Self::Avg { sum: s1, count: c1 }, Self::Avg { sum: s2, count: c2 }) => {
                *s1 += s2;
                *c1 += c2;
            }
            (Self::Collect(a), Self::Collect(b)) => a.extend_from_slice(b),
            (Self::CollectSet(a), Self::CollectSet(b)) | (Self::Distinct(a), Self::Distinct(b)) => {
                a.extend(b.iter().cloned())
            }
            (Self::Percentile { values: a, .. }, Self::Percentile { values: b, .. })
            | (Self::PercentileCont { values: a, .. }, Self::PercentileCont { values: b, .. }) => {
                a.extend_from_slice(b)
            }
            (Self::Median(a), Self::Median(b)) => a.extend_from_slice(b),
            (Self::Mode(a), Self::Mode(b)) => a.extend_from_slice(b),
            (
                Self::Std {
                    n: n1,
                    mean: m1,
                    m2: v1,
                },
                Self::Std {
                    n: n2,
                    mean: m2,
                    m2: v2,
                },
            )
            | (
                Self::StddevPop {
                    n: n1,
                    mean: m1,
                    m2: v1,
                },
                Self::StddevPop {
                    n: n2,
                    mean: m2,
                    m2: v2,
                },
            )
            | (
                Self::StddevSamp {
                    n: n1,
                    mean: m1,
                    m2: v1,
                },
                Self::StddevSamp {
                    n: n2,
                    mean: m2,
                    m2: v2,
                },
            )
            | (
                Self::Variance {
                    n: n1,
                    mean: m1,
                    m2: v1,
                },
                Self::Variance {
                    n: n2,
                    mean: m2,
                    m2: v2,
                },
            ) => welford_merge(n1, m1, v1, n2, m2, v2),
            (Self::Product(a), Self::Product(b)) => {
                if let (Some(x), Some(y)) = (*a, *b) {
                    *a = Some(x * y);
                } else if a.is_none() {
                    *a = *b;
                }
            }
            (Self::BitAnd(a), Self::BitAnd(b)) => {
                if let (Some(x), Some(y)) = (*a, *b) {
                    *a = Some(x & y);
                } else if a.is_none() {
                    *a = *b;
                }
            }
            (Self::BitOr(a), Self::BitOr(b)) => {
                if let (Some(x), Some(y)) = (*a, *b) {
                    *a = Some(x | y);
                } else if a.is_none() {
                    *a = *b;
                }
            }
            (Self::BoolAnd(a), Self::BoolAnd(b)) => {
                if let (Some(x), Some(y)) = (*a, *b) {
                    *a = Some(x && y);
                } else if a.is_none() {
                    *a = *b;
                }
            }
            (Self::BoolOr(a), Self::BoolOr(b)) => {
                if let (Some(x), Some(y)) = (*a, *b) {
                    *a = Some(x || y);
                } else if a.is_none() {
                    *a = *b;
                }
            }
            (Self::GroupConcat { parts: a, .. }, Self::GroupConcat { parts: b, .. })
            | (
                Self::GroupConcatWithOrder { parts: a, .. },
                Self::GroupConcatWithOrder { parts: b, .. },
            ) => a.extend_from_slice(b),
            (Self::VecSum(a), Self::VecSum(b)) => {
                if let (Some(x), Some(y)) = (a.as_mut(), b.as_ref()) {
                    if x.len() == y.len() {
                        for (xi, yi) in x.iter_mut().zip(y) {
                            *xi += yi;
                        }
                    }
                } else if a.is_none() {
                    *a = b.clone();
                }
            }
            (Self::VecAvg { sum: s1, count: c1 }, Self::VecAvg { sum: s2, count: c2 }) => {
                if s1.is_empty() {
                    *s1 = s2.clone();
                } else if s1.len() == s2.len() {
                    for (x, y) in s1.iter_mut().zip(s2) {
                        *x += y;
                    }
                }
                *c1 += c2;
            }
            _ => {}
        }
    }

    /// Produce the final aggregate Value from this accumulator.
    pub fn finalize(&self) -> Value {
        match self {
            Self::Count(count) => Value::BigInt(*count as i64),
            Self::Sum(sum) => integral_or_double(*sum),
            Self::Min(Some(v)) | Self::Max(Some(v)) => v.clone(),
            Self::Min(None) | Self::Max(None) => Value::Null(NullType::Null),
            Self::Avg { sum, count } => {
                if *count == 0 {
                    Value::Null(NullType::Null)
                } else {
                    Value::Double(*sum / *count as f64)
                }
            }
            Self::Collect(values) => {
                if values.is_empty() {
                    Value::Null(NullType::Null)
                } else {
                    let mut list = linkrs_core::value::List::new();
                    for v in values {
                        list.push(v.clone());
                    }
                    Value::List(Box::new(list))
                }
            }
            Self::CollectSet(set) | Self::Distinct(set) => {
                if set.is_empty() {
                    Value::Null(NullType::Null)
                } else {
                    Value::set(set.clone())
                }
            }
            Self::Percentile { values, percentile }
            | Self::PercentileCont { values, percentile } => percentile_of(values, *percentile),
            Self::Median(values) => median_of(values),
            Self::Mode(values) => mode_of(values),
            Self::Std { n, mean: _, m2 } | Self::Variance { n, mean: _, m2 } => {
                if *n == 0 {
                    Value::Null(NullType::Null)
                } else if matches!(self, Self::Std { .. }) {
                    Value::Double((m2 / *n as f64).sqrt())
                } else {
                    Value::Double(m2 / *n as f64)
                }
            }
            Self::StddevPop { n, mean: _, m2 } => {
                if *n == 0 {
                    Value::Null(NullType::Null)
                } else {
                    Value::Double((m2 / *n as f64).sqrt())
                }
            }
            Self::StddevSamp { n, mean: _, m2 } => {
                if *n < 2 {
                    Value::Null(NullType::Null)
                } else {
                    Value::Double((m2 / (*n - 1) as f64).sqrt())
                }
            }
            Self::Product(None) => Value::BigInt(0),
            Self::Product(Some(p)) => integral_or_double(*p),
            Self::BitAnd(Some(v)) | Self::BitOr(Some(v)) => Value::BigInt(*v),
            Self::BitAnd(None) | Self::BitOr(None) => Value::Null(NullType::Null),
            Self::BoolAnd(Some(v)) | Self::BoolOr(Some(v)) => Value::Bool(*v),
            Self::BoolAnd(None) | Self::BoolOr(None) => Value::Null(NullType::Null),
            Self::GroupConcat { parts, separator }
            | Self::GroupConcatWithOrder { parts, separator } => {
                Value::string(parts.join(separator))
            }
            Self::VecSum(Some(vec)) => Value::vector(vec.clone()),
            Self::VecSum(None) => Value::Null(NullType::NaN),
            Self::VecAvg { sum, count } => {
                if *count == 0 {
                    Value::Null(NullType::NaN)
                } else {
                    Value::vector(sum.iter().map(|x| x / *count as f32).collect())
                }
            }
        }
    }

    pub fn name(&self) -> &'static str {
        match self {
            Self::Count(_) => "Count",
            Self::Sum(_) => "Sum",
            Self::Min(_) => "Min",
            Self::Max(_) => "Max",
            Self::Avg { .. } => "Avg",
            Self::Collect(_) => "Collect",
            Self::CollectSet(_) => "CollectSet",
            Self::Distinct(_) => "Distinct",
            Self::Percentile { .. } => "Percentile",
            Self::PercentileCont { .. } => "PercentileCont",
            Self::Median(_) => "Median",
            Self::Mode(_) => "Mode",
            Self::Std { .. } => "Std",
            Self::StddevPop { .. } => "StddevPop",
            Self::StddevSamp { .. } => "StddevSamp",
            Self::Variance { .. } => "Variance",
            Self::Product(_) => "Product",
            Self::BitAnd(_) => "BitAnd",
            Self::BitOr(_) => "BitOr",
            Self::BoolAnd(_) => "BoolAnd",
            Self::BoolOr(_) => "BoolOr",
            Self::GroupConcat { .. } => "GroupConcat",
            Self::GroupConcatWithOrder { .. } => "GroupConcatWithOrder",
            Self::VecSum(_) => "VecSum",
            Self::VecAvg { .. } => "VecAvg",
        }
    }
}

fn numeric_value(value: &Value) -> Option<f64> {
    match value {
        Value::Int(n) => Some(*n as f64),
        Value::BigInt(n) => Some(*n as f64),
        Value::Float(f) => Some(*f as f64),
        Value::Double(f) => Some(*f),
        _ => None,
    }
}

fn integral_or_double(value: f64) -> Value {
    if value.fract() == 0.0 && value.is_finite() {
        Value::BigInt(value as i64)
    } else {
        Value::Double(value)
    }
}

fn welford_update(n: &mut u64, mean: &mut f64, m2: &mut f64, x: f64) {
    *n += 1;
    let delta = x - *mean;
    *mean += delta / *n as f64;
    let delta2 = x - *mean;
    *m2 += delta * delta2;
}

fn welford_merge(n1: &mut u64, mean1: &mut f64, m21: &mut f64, n2: &u64, mean2: &f64, m22: &f64) {
    if *n2 == 0 {
        return;
    }
    if *n1 == 0 {
        *n1 = *n2;
        *mean1 = *mean2;
        *m21 = *m22;
        return;
    }
    let total = *n1 + *n2;
    let delta = mean2 - *mean1;
    let new_mean = (*n1 as f64 * *mean1 + *n2 as f64 * *mean2) / total as f64;
    let new_m2 = *m21 + *m22 + delta * delta * *n1 as f64 * *n2 as f64 / total as f64;
    *n1 = total;
    *mean1 = new_mean;
    *m21 = new_m2;
}

fn percentile_of(values: &[f64], percentile: f64) -> Value {
    if values.is_empty() || !(0.0..=100.0).contains(&percentile) {
        return Value::Null(NullType::Null);
    }
    let mut sorted = values.to_vec();
    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
    let index = (percentile / 100.0) * (sorted.len() - 1) as f64;
    let lower = index.floor() as usize;
    let upper = index.ceil() as usize;
    if lower == upper {
        Value::Double(sorted[lower])
    } else {
        let weight = index - lower as f64;
        Value::Double(sorted[lower] + weight * (sorted[upper] - sorted[lower]))
    }
}

fn median_of(values: &[f64]) -> Value {
    if values.is_empty() {
        return Value::Null(NullType::Null);
    }
    let mut sorted = values.to_vec();
    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
    let len = sorted.len();
    if len.is_multiple_of(2) {
        Value::Double((sorted[len / 2 - 1] + sorted[len / 2]) / 2.0)
    } else {
        Value::Double(sorted[len / 2])
    }
}

fn mode_of(values: &[Value]) -> Value {
    if values.is_empty() {
        return Value::Null(NullType::Null);
    }
    let mut frequency: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
    for value in values {
        *frequency.entry(format!("{}", value)).or_insert(0) += 1;
    }
    let Some(mode_str) = frequency
        .into_iter()
        .max_by_key(|(_, count)| *count)
        .map(|(key, _)| key)
    else {
        return Value::Null(NullType::Null);
    };
    if let Ok(int_val) = mode_str.parse::<i32>() {
        Value::Int(int_val)
    } else if let Ok(float_val) = mode_str.parse::<f64>() {
        Value::Double(float_val)
    } else if mode_str == "true" {
        Value::Bool(true)
    } else if mode_str == "false" {
        Value::Bool(false)
    } else {
        Value::string(mode_str)
    }
}

/// Serialize an accumulator state into a Value for inter-operator transfer.
/// The partial aggregate output stores accumulator states as Value rows.
pub fn accumulator_to_value(acc: &AggregateAccumulator) -> Value {
    match acc {
        AggregateAccumulator::Count(c) => Value::BigInt(*c as i64),
        AggregateAccumulator::Sum(s) => Value::Double(*s),
        AggregateAccumulator::Min(Some(v)) | AggregateAccumulator::Max(Some(v)) => v.clone(),
        AggregateAccumulator::Min(None) | AggregateAccumulator::Max(None) => {
            Value::Null(NullType::Null)
        }
        AggregateAccumulator::Avg { sum, count } => {
            let mut list = linkrs_core::value::List::new();
            list.push(Value::Double(*sum));
            list.push(Value::BigInt(*count as i64));
            Value::List(Box::new(list))
        }
        AggregateAccumulator::Collect(values) => value_list_of_values(values),
        AggregateAccumulator::CollectSet(set) | AggregateAccumulator::Distinct(set) => {
            if set.is_empty() {
                Value::Null(NullType::Null)
            } else {
                Value::List(Box::new(linkrs_core::value::List::from(
                    set.iter().cloned().collect::<Vec<_>>(),
                )))
            }
        }
        AggregateAccumulator::Percentile { values, .. }
        | AggregateAccumulator::PercentileCont { values, .. }
        | AggregateAccumulator::Median(values) => value_list_of_f64(values),
        AggregateAccumulator::Mode(values) => value_list_of_values(values),
        AggregateAccumulator::Std { n, mean, m2 }
        | AggregateAccumulator::StddevPop { n, mean, m2 }
        | AggregateAccumulator::StddevSamp { n, mean, m2 }
        | AggregateAccumulator::Variance { n, mean, m2 } => {
            if *n == 0 {
                Value::Null(NullType::Null)
            } else {
                let mut list = linkrs_core::value::List::new();
                list.push(Value::BigInt(*n as i64));
                list.push(Value::Double(*mean));
                list.push(Value::Double(*m2));
                Value::List(Box::new(list))
            }
        }
        AggregateAccumulator::Product(Some(p)) => Value::Double(*p),
        AggregateAccumulator::Product(None) => Value::Null(NullType::Null),
        AggregateAccumulator::BitAnd(Some(v)) | AggregateAccumulator::BitOr(Some(v)) => {
            Value::BigInt(*v)
        }
        AggregateAccumulator::BitAnd(None) | AggregateAccumulator::BitOr(None) => {
            Value::Null(NullType::Null)
        }
        AggregateAccumulator::BoolAnd(Some(v)) | AggregateAccumulator::BoolOr(Some(v)) => {
            Value::Bool(*v)
        }
        AggregateAccumulator::BoolAnd(None) | AggregateAccumulator::BoolOr(None) => {
            Value::Null(NullType::Null)
        }
        AggregateAccumulator::GroupConcat { parts, .. }
        | AggregateAccumulator::GroupConcatWithOrder { parts, .. } => {
            if parts.is_empty() {
                Value::Null(NullType::Null)
            } else {
                Value::List(Box::new(linkrs_core::value::List::from(
                    parts.iter().cloned().map(Value::string).collect::<Vec<_>>(),
                )))
            }
        }
        AggregateAccumulator::VecSum(Some(vec)) => {
            value_list_of_f64(&vec.iter().map(|x| *x as f64).collect::<Vec<_>>())
        }
        AggregateAccumulator::VecSum(None) => Value::Null(NullType::Null),
        AggregateAccumulator::VecAvg { sum, count } => {
            if *count == 0 {
                Value::Null(NullType::Null)
            } else {
                let mut list = linkrs_core::value::List::new();
                let mut inner = linkrs_core::value::List::new();
                for x in sum {
                    inner.push(Value::Double(*x as f64));
                }
                list.push(Value::List(Box::new(inner)));
                list.push(Value::BigInt(*count as i64));
                Value::List(Box::new(list))
            }
        }
    }
}

fn value_list_of_values(values: &[Value]) -> Value {
    if values.is_empty() {
        Value::Null(NullType::Null)
    } else {
        Value::List(Box::new(linkrs_core::value::List::from(values.to_vec())))
    }
}

fn value_list_of_f64(values: &[f64]) -> Value {
    if values.is_empty() {
        Value::Null(NullType::Null)
    } else {
        Value::List(Box::new(linkrs_core::value::List::from(
            values.iter().map(|x| Value::Double(*x)).collect::<Vec<_>>(),
        )))
    }
}

/// Produce the accumulator final value from an encoded partial Value.
/// This reverses `accumulator_to_value` and calls `finalize`.
pub fn finalize_accumulator_value(
    func: &AggregateFunction,
    partial_value: &Value,
    _count_accumulator: Option<&AggregateAccumulator>,
) -> Value {
    match decode_partial(func, partial_value) {
        Some(acc) => acc.finalize(),
        None => partial_value.clone(),
    }
}

/// Decode a partial-accumulator Value back into an accumulator state.
/// This reverses `accumulator_to_value`.
pub fn decode_partial(func: &AggregateFunction, value: &Value) -> Option<AggregateAccumulator> {
    decode_partial_with_args(func, &[], value)
}

/// Decode a partial-accumulator Value back into an accumulator state,
/// using the given args to initialize parameters (separator, percentile, etc.).
pub fn decode_partial_with_args(
    func: &AggregateFunction,
    args: &[Expression],
    value: &Value,
) -> Option<AggregateAccumulator> {
    let mut acc = AggregateAccumulator::for_function(func, args)?;
    match &mut acc {
        AggregateAccumulator::Count(c) => {
            if let Value::BigInt(n) = value {
                *c = *n as u64;
            }
        }
        AggregateAccumulator::Sum(s) => {
            if let Value::Double(n) = value {
                *s = *n;
            } else if let Value::BigInt(n) = value {
                *s = *n as f64;
            }
        }
        AggregateAccumulator::Min(ref mut v) | AggregateAccumulator::Max(ref mut v) => {
            if !matches!(value, Value::Null(_)) {
                *v = Some(value.clone());
            }
        }
        AggregateAccumulator::Avg { sum, count } => {
            if let Value::List(list) = value {
                if list.len() >= 2 {
                    if let Value::Double(s) = &list.values[0] {
                        *sum = *s;
                    }
                    if let Value::BigInt(c) = &list.values[1] {
                        *count = *c as u64;
                    }
                }
            }
        }
        AggregateAccumulator::Collect(values) => {
            if let Value::List(list) = value {
                *values = list.values.clone();
            }
        }
        AggregateAccumulator::CollectSet(set) | AggregateAccumulator::Distinct(set) => {
            if let Value::List(list) = value {
                *set = list.values.iter().cloned().collect();
            }
        }
        AggregateAccumulator::Percentile { values, .. }
        | AggregateAccumulator::PercentileCont { values, .. }
        | AggregateAccumulator::Median(values) => {
            *values = f64_list_from_value(value);
        }
        AggregateAccumulator::Mode(values) => {
            if let Value::List(list) = value {
                *values = list.values.clone();
            }
        }
        AggregateAccumulator::Std { n, mean, m2 }
        | AggregateAccumulator::StddevPop { n, mean, m2 }
        | AggregateAccumulator::StddevSamp { n, mean, m2 }
        | AggregateAccumulator::Variance { n, mean, m2 } => {
            if let Value::List(list) = value {
                if list.len() >= 3 {
                    if let Value::BigInt(nn) = &list.values[0] {
                        *n = *nn as u64;
                    }
                    if let Value::Double(mm) = &list.values[1] {
                        *mean = *mm;
                    }
                    if let Value::Double(mm2) = &list.values[2] {
                        *m2 = *mm2;
                    }
                }
            }
        }
        AggregateAccumulator::Product(p) => match value {
            Value::Double(n) => *p = Some(*n),
            Value::BigInt(n) => *p = Some(*n as f64),
            _ => *p = None,
        },
        AggregateAccumulator::BitAnd(v) | AggregateAccumulator::BitOr(v) => {
            if let Value::BigInt(n) = value {
                *v = Some(*n);
            }
        }
        AggregateAccumulator::BoolAnd(v) | AggregateAccumulator::BoolOr(v) => {
            if let Value::Bool(b) = value {
                *v = Some(*b);
            }
        }
        AggregateAccumulator::GroupConcat { parts, .. }
        | AggregateAccumulator::GroupConcatWithOrder { parts, .. } => {
            if let Value::List(list) = value {
                *parts = list.values.iter().map(|v| format!("{}", v)).collect();
            }
        }
        AggregateAccumulator::VecSum(sum) => match value {
            Value::List(list) => {
                *sum = Some(
                    list.values
                        .iter()
                        .filter_map(|v| match v {
                            Value::Double(x) => Some(*x as f32),
                            Value::Int(x) => Some(*x as f32),
                            Value::BigInt(x) => Some(*x as f32),
                            _ => None,
                        })
                        .collect(),
                );
            }
            _ => *sum = None,
        },
        AggregateAccumulator::VecAvg { sum, count } => {
            if let Value::List(list) = value {
                if list.len() >= 2 {
                    if let Value::List(inner) = &list.values[0] {
                        *sum = inner
                            .values
                            .iter()
                            .filter_map(|v| match v {
                                Value::Double(x) => Some(*x as f32),
                                Value::Int(x) => Some(*x as f32),
                                Value::BigInt(x) => Some(*x as f32),
                                _ => None,
                            })
                            .collect();
                    }
                    if let Value::BigInt(c) = &list.values[1] {
                        *count = *c as u64;
                    }
                }
            }
        }
    }
    Some(acc)
}

fn f64_list_from_value(value: &Value) -> Vec<f64> {
    match value {
        Value::List(list) => list
            .values
            .iter()
            .filter_map(|v| match v {
                Value::Double(x) => Some(*x),
                Value::Int(x) => Some(*x as f64),
                Value::BigInt(x) => Some(*x as f64),
                _ => None,
            })
            .collect(),
        _ => Vec::new(),
    }
}

#[cfg(test)]
mod tests;