[pull] main from apache:main

datafusion/functions-aggregate/src/correlation.rs

@@ -281,6 +281,10 @@ impl Accumulator for CorrelationAccumulator {
         self.stddev2.retract_batch(&values[1..2])?;
         Ok(())
     }
+
+    fn supports_retract_batch(&self) -> bool {
+        true
+    }
 }
 
 #[derive(Default)]

datafusion/functions-aggregate/src/covariance.rs

@@ -305,6 +305,14 @@ impl Accumulator for CovarianceAccumulator {
                 _ => continue,
             };
 
+            if self.count <= 1 {
+                self.count = 0;
+                self.mean1 = 0.0;
+                self.mean2 = 0.0;
+                self.algo_const = 0.0;
+                continue;
+            }
+
             let new_count = self.count - 1;
             let delta1 = self.mean1 - value1;
             let new_mean1 = delta1 / new_count as f64 + self.mean1;
@@ -373,4 +381,8 @@ impl Accumulator for CovarianceAccumulator {
     fn size(&self) -> usize {
         size_of_val(self)
     }
+
+    fn supports_retract_batch(&self) -> bool {
+        true
+    }
 }

datafusion/functions-aggregate/src/variance.rs

@@ -348,6 +348,13 @@ impl Accumulator for VarianceAccumulator {
     fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
         let arr = as_float64_array(&values[0])?;
         for value in arr.iter().flatten() {
+            if self.count <= 1 {
+                self.count = 0;
+                self.mean = 0.0;
+                self.m2 = 0.0;
+                continue;
+            }
+
             let new_count = self.count - 1;
             let delta1 = self.mean - value;
             let new_mean = delta1 / new_count as f64 + self.mean;