surrealdb-core 3.3.1

//! Math aggregate functions.
//!
//! Provides aggregates for mathematical operations: sum, mean, min, max,
//! stddev, variance, and median.

use anyhow::Result;

use crate::catalog::aggregation;
use crate::exec::function::{Accumulator, AggregateFunction, Signature};
use crate::expr::Kind;
use crate::val::{Number, TryAdd as _, Value};

// ============================================================================
// Sum
// ============================================================================

/// math::sum - sums numeric values
#[derive(Debug, Clone, Copy, Default)]
pub struct MathSum;

impl AggregateFunction for MathSum {
	fn name(&self) -> &'static str {
		"math::sum"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(SumAccumulator::default())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

#[derive(Debug, Clone, Default)]
struct SumAccumulator {
	sum: Number,
}

impl Accumulator for SumAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			self.sum = self.sum + n;
		}
		// Skip non-numbers (matches old behavior)
		Ok(())
	}

	fn update_batch(&mut self, values: &[Value]) -> Result<()> {
		for value in values {
			if let Value::Number(n) = value {
				self.sum = self.sum + *n;
			}
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<SumAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		self.sum = self.sum + other.sum;
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		Ok(Value::Number(self.sum))
	}

	fn reset(&mut self) {
		self.sum = Number::Int(0);
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Mean
// ============================================================================

/// math::mean - calculates average of numeric values
#[derive(Debug, Clone, Copy, Default)]
pub struct MathMean;

impl AggregateFunction for MathMean {
	fn name(&self) -> &'static str {
		"math::mean"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(MeanAccumulator::default())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

#[derive(Debug, Clone, Default)]
struct MeanAccumulator {
	sum: Number,
	count: i64,
}

impl Accumulator for MeanAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			self.sum = self.sum + n;
			self.count += 1;
		}
		// Skip non-numbers
		Ok(())
	}

	fn update_batch(&mut self, values: &[Value]) -> Result<()> {
		for value in values {
			if let Value::Number(n) = value {
				self.sum = self.sum + *n;
				self.count += 1;
			}
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<MeanAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		self.sum = self.sum + other.sum;
		self.count += other.count;
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		if self.count == 0 {
			// Return NaN for empty groups (matches old behavior)
			Ok(Value::Number(Number::Float(f64::NAN)))
		} else {
			let mean = self.sum.to_float() / self.count as f64;
			Ok(Value::Number(Number::Float(mean)))
		}
	}

	fn reset(&mut self) {
		self.sum = Number::Int(0);
		self.count = 0;
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Min
// ============================================================================

/// math::min - finds minimum numeric value
#[derive(Debug, Clone, Copy, Default)]
pub struct MathMin;

impl AggregateFunction for MathMin {
	fn name(&self) -> &'static str {
		"math::min"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(MinAccumulator::default())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

#[derive(Debug, Clone, Default)]
struct MinAccumulator {
	min: Option<Number>,
}

impl Accumulator for MinAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			self.min = Some(match &self.min {
				None => n,
				Some(current) => {
					if n < *current {
						n
					} else {
						*current
					}
				}
			});
		}
		Ok(())
	}

	fn update_batch(&mut self, values: &[Value]) -> Result<()> {
		for value in values {
			if let Value::Number(n) = value {
				self.min = Some(match &self.min {
					None => *n,
					Some(current) => {
						if *n < *current {
							*n
						} else {
							*current
						}
					}
				});
			}
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<MinAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		if let Some(other_min) = &other.min {
			self.min = Some(match &self.min {
				None => *other_min,
				Some(current) => {
					if *other_min < *current {
						*other_min
					} else {
						*current
					}
				}
			});
		}
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		match &self.min {
			Some(n) => Ok(Value::Number(*n)),
			None => Ok(Value::Number(Number::Float(f64::INFINITY))),
		}
	}

	fn reset(&mut self) {
		self.min = None;
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Max
// ============================================================================

/// math::max - finds maximum numeric value
#[derive(Debug, Clone, Copy, Default)]
pub struct MathMax;

impl AggregateFunction for MathMax {
	fn name(&self) -> &'static str {
		"math::max"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(MaxAccumulator::default())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

#[derive(Debug, Clone, Default)]
struct MaxAccumulator {
	max: Option<Number>,
}

impl Accumulator for MaxAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			self.max = Some(match &self.max {
				None => n,
				Some(current) => {
					if n > *current {
						n
					} else {
						*current
					}
				}
			});
		}
		Ok(())
	}

	fn update_batch(&mut self, values: &[Value]) -> Result<()> {
		for value in values {
			if let Value::Number(n) = value {
				self.max = Some(match &self.max {
					None => *n,
					Some(current) => {
						if *n > *current {
							*n
						} else {
							*current
						}
					}
				});
			}
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<MaxAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		if let Some(other_max) = &other.max {
			self.max = Some(match &self.max {
				None => *other_max,
				Some(current) => {
					if *other_max > *current {
						*other_max
					} else {
						*current
					}
				}
			});
		}
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		match &self.max {
			Some(n) => Ok(Value::Number(*n)),
			None => Ok(Value::Number(Number::Float(f64::NEG_INFINITY))),
		}
	}

	fn reset(&mut self) {
		self.max = None;
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Stddev / Variance (shifted sum-of-squares)
// ============================================================================

/// math::stddev - sample standard deviation.
#[derive(Debug, Clone, Copy, Default)]
pub struct MathStddev;

impl AggregateFunction for MathStddev {
	fn name(&self) -> &'static str {
		"math::stddev"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(SpreadAccumulator::deviation())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

/// math::variance - sample variance.
#[derive(Debug, Clone, Copy, Default)]
pub struct MathVariance;

impl AggregateFunction for MathVariance {
	fn name(&self) -> &'static str {
		"math::variance"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(SpreadAccumulator::variance())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

/// Whether a [`SpreadAccumulator`] finalises to a variance or its square root.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Spread {
	Variance,
	Deviation,
}

/// Shared accumulator for `math::variance` and `math::stddev`.
///
/// Delegates to [`aggregation::shifted_accumulate`] and
/// [`aggregation::shifted_sample_variance`], the same functions the legacy
/// `GROUP BY` collector and materialized-view maintenance use, so all three
/// paths produce bit-identical results over the same input order. Arithmetic
/// stays in `Number`, so a group of `Decimal` inputs finalises as `Decimal`.
#[derive(Debug, Clone)]
struct SpreadAccumulator {
	spread: Spread,
	shift: Number,
	sum: Number,
	sum_of_squares: Number,
	count: i64,
}

impl SpreadAccumulator {
	fn new(spread: Spread) -> Self {
		Self {
			spread,
			shift: 0.0.into(),
			sum: 0.0.into(),
			sum_of_squares: 0.0.into(),
			count: 0,
		}
	}

	fn variance() -> Self {
		Self::new(Spread::Variance)
	}

	fn deviation() -> Self {
		Self::new(Spread::Deviation)
	}
}

impl Accumulator for SpreadAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			aggregation::shifted_accumulate(
				&mut self.shift,
				&mut self.sum,
				&mut self.sum_of_squares,
				&mut self.count,
				n,
			)?;
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<SpreadAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		// Both functions share this accumulator, so the downcast above no longer
		// distinguishes them the way the two separate Welford types used to.
		// Finalising against the wrong one would return a variance where a
		// deviation was asked for, silently.
		anyhow::ensure!(
			self.spread == other.spread,
			"Cannot merge a {:?} accumulator into a {:?} one",
			other.spread,
			self.spread
		);
		if other.count == 0 {
			return Ok(());
		}
		if self.count == 0 {
			self.shift = other.shift;
			self.sum = other.sum;
			self.sum_of_squares = other.sum_of_squares;
			self.count = other.count;
			return Ok(());
		}

		// The two halves were shifted against different assumed means, so
		// re-base `other` onto ours before adding.
		let (rebased_sum, rebased_sum_of_squares) = aggregation::shifted_rebase(
			other.sum,
			other.sum_of_squares,
			other.count,
			other.shift,
			self.shift,
		)?;

		self.sum = self.sum.try_add(rebased_sum)?;
		self.sum_of_squares = self.sum_of_squares.try_add(rebased_sum_of_squares)?;
		self.count += other.count;
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		let num = match self.spread {
			Spread::Variance => {
				aggregation::shifted_sample_variance(self.sum, self.sum_of_squares, self.count)
			}
			Spread::Deviation => {
				aggregation::shifted_sample_deviation(self.sum, self.sum_of_squares, self.count)
			}
		};
		Ok(Value::Number(num))
	}

	fn reset(&mut self) {
		*self = Self::new(self.spread);
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Median
// ============================================================================

/// math::median - calculates median of values
#[derive(Debug, Clone, Copy, Default)]
pub struct MathMedian;

impl AggregateFunction for MathMedian {
	fn name(&self) -> &'static str {
		"math::median"
	}

	fn create_accumulator(&self) -> Box<dyn Accumulator> {
		Box::new(MedianAccumulator::default())
	}

	fn signature(&self) -> Signature {
		Signature::new().arg("value", Kind::Number).returns(Kind::Number)
	}
}

#[derive(Debug, Clone, Default)]
struct MedianAccumulator {
	values: Vec<Number>,
}

impl Accumulator for MedianAccumulator {
	fn update(&mut self, value: Value) -> Result<()> {
		if let Value::Number(n) = value {
			self.values.push(n);
		}
		Ok(())
	}

	fn merge(&mut self, other: Box<dyn Accumulator>) -> Result<()> {
		let other = other
			.as_any()
			.downcast_ref::<MedianAccumulator>()
			.ok_or_else(|| anyhow::anyhow!("Cannot merge incompatible accumulators"))?;
		self.values.extend(other.values.iter().copied());
		Ok(())
	}

	fn finalize(&self) -> Result<Value> {
		if self.values.is_empty() {
			return Ok(Value::None);
		}

		// Sort the values
		let mut sorted = self.values.clone();
		sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));

		let len = sorted.len();
		let median = if len.is_multiple_of(2) {
			// Even number of elements: average of the two middle values
			let mid = len / 2;
			let a = sorted[mid - 1].to_float();
			let b = sorted[mid].to_float();
			Number::Float((a + b) / 2.0)
		} else {
			// Odd number of elements: middle value
			sorted[len / 2]
		};

		Ok(Value::Number(median))
	}

	fn reset(&mut self) {
		self.values.clear();
	}

	fn clone_box(&self) -> Box<dyn Accumulator> {
		Box::new(self.clone())
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
	use surrealdb_strand::Strand;

	use super::*;

	// Helper to extract f64 from Value
	fn as_float(v: &Value) -> f64 {
		match v {
			Value::Number(Number::Float(f)) => *f,
			Value::Number(Number::Int(i)) => *i as f64,
			_ => panic!("Expected Number, got {:?}", v),
		}
	}

	// Helper to check approximate equality for floats
	fn approx_eq(a: f64, b: f64, epsilon: f64) -> bool {
		(a - b).abs() < epsilon
	}

	// -------------------------------------------------------------------------
	// Sum tests
	// -------------------------------------------------------------------------

	#[test]
	fn sum_zero_items() {
		let func = MathSum;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 0.0);
	}

	#[test]
	fn sum_single_item() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 42.0);
	}

	#[test]
	fn sum_multiple_items() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		acc.update(Value::Number(Number::Int(3))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 6.0);
	}

	#[test]
	fn sum_merge() {
		let func = MathSum;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(1))).unwrap();
		acc1.update(Value::Number(Number::Int(2))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(3))).unwrap();
		acc2.update(Value::Number(Number::Int(4))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert_eq!(as_float(&result), 10.0);
	}

	// -------------------------------------------------------------------------
	// Mean tests
	// -------------------------------------------------------------------------

	#[test]
	fn mean_zero_items() {
		let func = MathMean;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert!(as_float(&result).is_nan());
	}

	#[test]
	fn mean_single_item() {
		let func = MathMean;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 42.0);
	}

	#[test]
	fn mean_multiple_items() {
		let func = MathMean;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		acc.update(Value::Number(Number::Int(4))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 3.0);
	}

	#[test]
	fn mean_merge() {
		let func = MathMean;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(2))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(4))).unwrap();
		acc2.update(Value::Number(Number::Int(6))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert_eq!(as_float(&result), 4.0);
	}

	// -------------------------------------------------------------------------
	// Min tests
	// -------------------------------------------------------------------------

	#[test]
	fn min_zero_items() {
		let func = MathMin;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), f64::INFINITY);
	}

	#[test]
	fn min_single_item() {
		let func = MathMin;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 42.0);
	}

	#[test]
	fn min_multiple_items() {
		let func = MathMin;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(3))).unwrap();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 1.0);
	}

	#[test]
	fn min_merge() {
		let func = MathMin;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(5))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(2))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert_eq!(as_float(&result), 2.0);
	}

	// -------------------------------------------------------------------------
	// Max tests
	// -------------------------------------------------------------------------

	#[test]
	fn max_zero_items() {
		let func = MathMax;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), f64::NEG_INFINITY);
	}

	#[test]
	fn max_single_item() {
		let func = MathMax;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 42.0);
	}

	#[test]
	fn max_multiple_items() {
		let func = MathMax;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		acc.update(Value::Number(Number::Int(3))).unwrap();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 3.0);
	}

	#[test]
	fn max_merge() {
		let func = MathMax;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(5))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(10))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert_eq!(as_float(&result), 10.0);
	}

	// -------------------------------------------------------------------------
	// Stddev tests
	// -------------------------------------------------------------------------

	#[test]
	fn stddev_zero_items() {
		// Empty group has no defined deviation; matches scalar `math::stddev`.
		let func = MathStddev;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert!(as_float(&result).is_nan());
	}

	#[test]
	fn stddev_single_item() {
		let func = MathStddev;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 0.0);
	}

	#[test]
	fn stddev_multiple_items() {
		let func = MathStddev;
		let mut acc = func.create_accumulator();
		// Values: 2, 4, 6 -> mean = 4, variance = 4, stddev = 2
		acc.update(Value::Number(Number::Int(2))).unwrap();
		acc.update(Value::Number(Number::Int(4))).unwrap();
		acc.update(Value::Number(Number::Int(6))).unwrap();
		let result = acc.finalize().unwrap();
		assert!(approx_eq(as_float(&result), 2.0, 1e-10));
	}

	#[test]
	fn stddev_merge() {
		let func = MathStddev;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(2))).unwrap();
		acc1.update(Value::Number(Number::Int(4))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(6))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert!(approx_eq(as_float(&result), 2.0, 1e-10));
	}

	#[test]
	fn stddev_numerical_stability() {
		// Test with large numbers that would cause issues with naive algorithm
		let func = MathStddev;
		let mut acc = func.create_accumulator();
		let base = 1e9;
		acc.update(Value::Number(Number::Float(base + 1.0))).unwrap();
		acc.update(Value::Number(Number::Float(base + 2.0))).unwrap();
		acc.update(Value::Number(Number::Float(base + 3.0))).unwrap();
		let result = acc.finalize().unwrap();
		// stddev of [1, 2, 3] offset by 1e9 should still be 1.0
		assert!(approx_eq(as_float(&result), 1.0, 1e-6));
	}

	// -------------------------------------------------------------------------
	// Variance tests
	// -------------------------------------------------------------------------

	#[test]
	fn variance_zero_items() {
		// Empty group has no defined variance; matches scalar `math::variance`.
		let func = MathVariance;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert!(as_float(&result).is_nan());
	}

	#[test]
	fn variance_single_item() {
		let func = MathVariance;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 0.0);
	}

	#[test]
	fn variance_multiple_items() {
		let func = MathVariance;
		let mut acc = func.create_accumulator();
		// Values: 2, 4, 6 -> mean = 4, variance = 4
		acc.update(Value::Number(Number::Int(2))).unwrap();
		acc.update(Value::Number(Number::Int(4))).unwrap();
		acc.update(Value::Number(Number::Int(6))).unwrap();
		let result = acc.finalize().unwrap();
		assert!(approx_eq(as_float(&result), 4.0, 1e-10));
	}

	#[test]
	fn variance_merge() {
		let func = MathVariance;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(2))).unwrap();
		acc1.update(Value::Number(Number::Int(4))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(6))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		assert!(approx_eq(as_float(&result), 4.0, 1e-10));
	}

	// -------------------------------------------------------------------------
	// Median tests
	// -------------------------------------------------------------------------

	#[test]
	fn median_zero_items() {
		let func = MathMedian;
		let acc = func.create_accumulator();
		let result = acc.finalize().unwrap();
		assert!(matches!(result, Value::None));
	}

	#[test]
	fn median_single_item() {
		let func = MathMedian;
		let mut acc = func.create_accumulator();
		acc.update(Value::Number(Number::Int(42))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 42.0);
	}

	#[test]
	fn median_odd_count() {
		let func = MathMedian;
		let mut acc = func.create_accumulator();
		// Values: 3, 1, 2 -> sorted: 1, 2, 3 -> median = 2
		acc.update(Value::Number(Number::Int(3))).unwrap();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 2.0);
	}

	#[test]
	fn median_even_count() {
		let func = MathMedian;
		let mut acc = func.create_accumulator();
		// Values: 4, 1, 3, 2 -> sorted: 1, 2, 3, 4 -> median = (2+3)/2 = 2.5
		acc.update(Value::Number(Number::Int(4))).unwrap();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		acc.update(Value::Number(Number::Int(3))).unwrap();
		acc.update(Value::Number(Number::Int(2))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 2.5);
	}

	#[test]
	fn median_merge() {
		let func = MathMedian;
		let mut acc1 = func.create_accumulator();
		acc1.update(Value::Number(Number::Int(1))).unwrap();
		acc1.update(Value::Number(Number::Int(5))).unwrap();

		let mut acc2 = func.create_accumulator();
		acc2.update(Value::Number(Number::Int(3))).unwrap();

		acc1.merge(acc2).unwrap();
		let result = acc1.finalize().unwrap();
		// [1, 5, 3] -> sorted [1, 3, 5] -> median = 3
		assert_eq!(as_float(&result), 3.0);
	}

	// -------------------------------------------------------------------------
	// update_batch tests - Sum
	// -------------------------------------------------------------------------

	#[test]
	fn sum_batch_empty() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		acc.update_batch(&[]).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 0.0);
	}

	#[test]
	fn sum_batch_multiple() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		let values = vec![
			Value::Number(Number::Int(1)),
			Value::Number(Number::Int(2)),
			Value::Number(Number::Int(3)),
		];
		acc.update_batch(&values).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 6.0);
	}

	#[test]
	fn sum_batch_skips_non_numbers() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		let values = vec![
			Value::Number(Number::Int(5)),
			Value::None,
			Value::String(Strand::new_static("test")),
			Value::Number(Number::Int(3)),
		];
		acc.update_batch(&values).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 8.0);
	}

	#[test]
	fn sum_batch_then_single() {
		let func = MathSum;
		let mut acc = func.create_accumulator();
		let values = vec![Value::Number(Number::Int(1)), Value::Number(Number::Int(2))];
		acc.update_batch(&values).unwrap();
		acc.update(Value::Number(Number::Int(7))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 10.0);
	}

	// -------------------------------------------------------------------------
	// update_batch tests - Mean
	// -------------------------------------------------------------------------

	#[test]
	fn mean_batch_empty() {
		let func = MathMean;
		let mut acc = func.create_accumulator();
		acc.update_batch(&[]).unwrap();
		let result = acc.finalize().unwrap();
		assert!(as_float(&result).is_nan());
	}

	#[test]
	fn mean_batch_multiple() {
		let func = MathMean;
		let mut acc = func.create_accumulator();
		let values = vec![Value::Number(Number::Int(2)), Value::Number(Number::Int(4))];
		acc.update_batch(&values).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 3.0);
	}

	#[test]
	fn mean_batch_then_single() {
		let func = MathMean;
		let mut acc = func.create_accumulator();
		let values = vec![Value::Number(Number::Int(2))];
		acc.update_batch(&values).unwrap();
		acc.update(Value::Number(Number::Int(4))).unwrap();
		acc.update(Value::Number(Number::Int(6))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 4.0);
	}

	// -------------------------------------------------------------------------
	// update_batch tests - Min
	// -------------------------------------------------------------------------

	#[test]
	fn min_batch_empty() {
		let func = MathMin;
		let mut acc = func.create_accumulator();
		acc.update_batch(&[]).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), f64::INFINITY);
	}

	#[test]
	fn min_batch_multiple() {
		let func = MathMin;
		let mut acc = func.create_accumulator();
		let values = vec![
			Value::Number(Number::Int(3)),
			Value::Number(Number::Int(1)),
			Value::Number(Number::Int(2)),
		];
		acc.update_batch(&values).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 1.0);
	}

	#[test]
	fn min_batch_then_single() {
		let func = MathMin;
		let mut acc = func.create_accumulator();
		let values = vec![Value::Number(Number::Int(5)), Value::Number(Number::Int(3))];
		acc.update_batch(&values).unwrap();
		acc.update(Value::Number(Number::Int(1))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 1.0);
	}

	// -------------------------------------------------------------------------
	// update_batch tests - Max
	// -------------------------------------------------------------------------

	#[test]
	fn max_batch_empty() {
		let func = MathMax;
		let mut acc = func.create_accumulator();
		acc.update_batch(&[]).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), f64::NEG_INFINITY);
	}

	#[test]
	fn max_batch_multiple() {
		let func = MathMax;
		let mut acc = func.create_accumulator();
		let values = vec![
			Value::Number(Number::Int(1)),
			Value::Number(Number::Int(3)),
			Value::Number(Number::Int(2)),
		];
		acc.update_batch(&values).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 3.0);
	}

	#[test]
	fn max_batch_then_single() {
		let func = MathMax;
		let mut acc = func.create_accumulator();
		let values = vec![Value::Number(Number::Int(5)), Value::Number(Number::Int(3))];
		acc.update_batch(&values).unwrap();
		acc.update(Value::Number(Number::Int(10))).unwrap();
		let result = acc.finalize().unwrap();
		assert_eq!(as_float(&result), 10.0);
	}
}