Should fail VAR(DISTINCT) but doesn't

Author: yyin-dev

datafusion/functions-aggregate/src/variance.rs

@@ -28,7 +28,6 @@ use arrow::{
 use datafusion_common::{downcast_value, plan_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::{
     function::{AccumulatorArgs, StateFieldsArgs},
-    type_coercion::aggregates::NUMERICS,
     utils::format_state_name,
     Accumulator, AggregateUDFImpl, Signature, Volatility,
 };
@@ -65,8 +64,8 @@ impl Default for VarianceSample {
 impl VarianceSample {
     pub fn new() -> Self {
         Self {
-            aliases: vec![String::from("var")],
-            signature: Signature::uniform(2, NUMERICS.to_vec(), Volatility::Immutable),
+            aliases: vec![String::from("var_sample")],
+            signature: Signature::numeric(1, Volatility::Immutable),
         }
     }
 }
@@ -251,4 +250,8 @@ impl Accumulator for VarianceAccumulator {
     fn size(&self) -> usize {
         std::mem::size_of_val(self)
     }
+
+    fn supports_retract_batch(&self) -> bool {
+        true
+    }
 }

datafusion/physical-expr/src/aggregate/build_in.rs

@@ -742,8 +742,6 @@ mod tests {
         Ok(())
     }
 
-    // TODO (yyin): Add back test
-
     #[test]
     fn test_var_pop_expr() -> Result<()> {
         let funcs = vec![AggregateFunction::VariancePop];
@@ -1012,8 +1010,6 @@ mod tests {
         assert!(observed.is_err());
     }
 
-    // TODO (yyin): Add back tests to sqllogictest
-
     #[test]
     fn test_stddev_return_type() -> Result<()> {
         let observed = AggregateFunction::Stddev.return_type(&[DataType::Float32])?;

datafusion/physical-expr/src/aggregate/variance.rs

@@ -35,88 +35,13 @@ use datafusion_common::downcast_value;
 use datafusion_common::{DataFusionError, Result, ScalarValue};
 use datafusion_expr::Accumulator;
 
-/// VAR and VAR_SAMP aggregate expression
-#[derive(Debug)]
-pub struct Variance {
-    name: String,
-    expr: Arc<dyn PhysicalExpr>,
-}
-
 /// VAR_POP aggregate expression
 #[derive(Debug)]
 pub struct VariancePop {
     name: String,
     expr: Arc<dyn PhysicalExpr>,
 }
 
-impl Variance {
-    /// Create a new VARIANCE aggregate function
-    pub fn new(
-        expr: Arc<dyn PhysicalExpr>,
-        name: impl Into<String>,
-        data_type: DataType,
-    ) -> Self {
-        // the result of variance just support FLOAT64 data type.
-        assert!(matches!(data_type, DataType::Float64));
-        Self {
-            name: name.into(),
-            expr,
-        }
-    }
-}
-
-impl AggregateExpr for Variance {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn field(&self) -> Result<Field> {
-        Ok(Field::new(&self.name, DataType::Float64, true))
-    }
-
-    fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
-        Ok(Box::new(VarianceAccumulator::try_new(StatsType::Sample)?))
-    }
-
-    fn create_sliding_accumulator(&self) -> Result<Box<dyn Accumulator>> {
-        Ok(Box::new(VarianceAccumulator::try_new(StatsType::Sample)?))
-    }
-
-    fn state_fields(&self) -> Result<Vec<Field>> {
-        Ok(vec![
-            Field::new(
-                format_state_name(&self.name, "count"),
-                DataType::UInt64,
-                true,
-            ),
-            Field::new(
-                format_state_name(&self.name, "mean"),
-                DataType::Float64,
-                true,
-            ),
-            Field::new(format_state_name(&self.name, "m2"), DataType::Float64, true),
-        ])
-    }
-
-    fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
-        vec![self.expr.clone()]
-    }
-
-    fn name(&self) -> &str {
-        &self.name
-    }
-}
-
-impl PartialEq<dyn Any> for Variance {
-    fn eq(&self, other: &dyn Any) -> bool {
-        down_cast_any_ref(other)
-            .downcast_ref::<Self>()
-            .map(|x| self.name == x.name && self.expr.eq(&x.expr))
-            .unwrap_or(false)
-    }
-}
-
 impl VariancePop {
     /// Create a new VAR_POP aggregate function
     pub fn new(

datafusion/physical-expr/src/expressions/mod.rs

@@ -60,7 +60,7 @@ pub use crate::aggregate::stddev::{Stddev, StddevPop};
 pub use crate::aggregate::string_agg::StringAgg;
 pub use crate::aggregate::sum::Sum;
 pub use crate::aggregate::sum_distinct::DistinctSum;
-pub use crate::aggregate::variance::{Variance, VariancePop};
+pub use crate::aggregate::variance::VariancePop;
 pub use crate::window::cume_dist::{cume_dist, CumeDist};
 pub use crate::window::lead_lag::{lag, lead, WindowShift};
 pub use crate::window::nth_value::NthValue;

datafusion/sqllogictest/Cargo.toml

@@ -40,7 +40,7 @@ bigdecimal = { workspace = true }
 bytes = { workspace = true, optional = true }
 chrono = { workspace = true, optional = true }
 clap = { version = "4.4.8", features = ["derive", "env"] }
-datafusion = { workspace = true, default-features = true }
+datafusion = { workspace = true, default-features = true, features = ["avro"] }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-common-runtime = { workspace = true, default-features = true }
 futures = { workspace = true }
@@ -60,7 +60,13 @@ tokio-postgres = { version = "0.7.7", optional = true }
 
 [features]
 avro = ["datafusion/avro"]
-postgres = ["bytes", "chrono", "tokio-postgres", "postgres-types", "postgres-protocol"]
+postgres = [
+    "bytes",
+    "chrono",
+    "tokio-postgres",
+    "postgres-types",
+    "postgres-protocol",
+]
 
 [dev-dependencies]
 env_logger = { workspace = true }