feat: implement removeInput method for regression, covariance, and correlation accumulators

Author: Cool6689

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java

@@ -5669,6 +5669,54 @@ public void statFunctionsInHavingTest() {
         DATABASE_NAME);
   }
 
+  @Test
+  public void statFunctionsWindowFrameTest() {
+    String[] expectedHeader =
+        new String[] {
+          "device_id", "time", "corr_v", "covar_pop_v", "covar_samp_v", "slope_v", "intercept_v"
+        };
+    String[] retArray =
+        new String[] {
+          "d1,1970-01-01T00:00:00.001Z,1.0,0.25,0.5,1.0,0.0,",
+          "d1,1970-01-01T00:00:00.002Z,0.8660254037844386,0.3333333333333333,0.49999999999999994,1.5,-0.5,",
+          "d1,1970-01-01T00:00:00.003Z,-0.8660254037844385,-0.3333333333333333,-0.5,-1.4999999999999998,5.5,",
+          "d1,1970-01-01T00:00:00.004Z,-0.8660254037844385,-0.3333333333333333,-0.49999999999999994,-1.4999999999999998,6.0,",
+          "d1,1970-01-01T00:00:00.005Z,null,0.0,0.0,null,null,",
+        };
+    tableResultSetEqualTest(
+        "select device_id, time, "
+            + "corr(s1, s2) over (partition by device_id order by time rows between 1 preceding and 1 following) as corr_v, "
+            + "covar_pop(s1, s2) over (partition by device_id order by time rows between 1 preceding and 1 following) as covar_pop_v, "
+            + "covar_samp(s1, s2) over (partition by device_id order by time rows between 1 preceding and 1 following) as covar_samp_v, "
+            + "regr_slope(s1, s2) over (partition by device_id order by time rows between 1 preceding and 1 following) as slope_v, "
+            + "regr_intercept(s1, s2) over (partition by device_id order by time rows between 1 preceding and 1 following) as intercept_v "
+            + "from stat_table where device_id = 'd1' order by device_id, time",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void statMomentsWindowFrameTest() {
+    String[] expectedHeader = new String[] {"device_id", "time", "skew_v", "kurt_v"};
+    String[] retArray =
+        new String[] {
+          "d1,1970-01-01T00:00:00.001Z,null,null,",
+          "d1,1970-01-01T00:00:00.002Z,null,null,",
+          "d1,1970-01-01T00:00:00.003Z,0.0,null,",
+          "d1,1970-01-01T00:00:00.004Z,0.0,-1.200000000000001,",
+          "d1,1970-01-01T00:00:00.005Z,0.0,-1.200000000000001,",
+        };
+    tableResultSetEqualTest(
+        "select device_id, time, "
+            + "skewness(s1) over (partition by device_id order by time rows between 3 preceding and current row) as skew_v, "
+            + "kurtosis(s1) over (partition by device_id order by time rows between 3 preceding and current row) as kurt_v "
+            + "from stat_table where device_id = 'd1' order by device_id, time",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
   @Test
   public void statFunctionsErrorTest() {
     String typeError = "only support numeric data types [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]";

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCentralMomentAccumulator.java

@@ -36,6 +36,7 @@ public class TableCentralMomentAccumulator implements TableAccumulator {
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(TableCentralMomentAccumulator.class);
   private static final int INTERMEDIATE_SIZE = Long.BYTES + 4 * Double.BYTES;
+  private static final double EPSILON = 1e-12;
 
   private final TSDataType seriesDataType;
   private final CentralMomentAccumulator.MomentType momentType;
@@ -229,7 +230,45 @@ public TableAccumulator copy() {
 
   @Override
   public void removeInput(Column[] arguments) {
-    throw new UnsupportedOperationException();
+    checkArgument(arguments.length == 1, "Input of CentralMoment should be 1");
+    if (count == 0 || arguments[0].isNull(0)) {
+      return;
+    }
+
+    double value = getDoubleValue(arguments[0], 0);
+    if (count == 1) {
+      reset();
+      return;
+    }
+
+    long nTotal = count;
+    long nA = nTotal - 1;
+
+    double meanA = (nTotal * mean - value) / nA;
+
+    double delta = value - meanA;
+    double delta2 = delta * delta;
+    double delta3 = delta * delta2;
+    double delta4 = delta2 * delta2;
+
+    double m2A = m2 - delta2 * nA / nTotal;
+    double m3A =
+        m3 - delta3 * nA * (nA - 1) / ((double) nTotal * nTotal) + 3.0 * delta * m2A / nTotal;
+    double m4A =
+        m4
+            - delta4 * nA * ((double) nA * nA - nA + 1) / ((double) nTotal * nTotal * nTotal)
+            - 6.0 * delta2 * m2A / ((double) nTotal * nTotal)
+            + 4.0 * delta * m3A / nTotal;
+
+    count = nA;
+    mean = meanA;
+    m2 = normalizeZero(m2A);
+    m3 = normalizeZero(m3A);
+    m4 = normalizeZero(m4A);
+  }
+
+  private double normalizeZero(double value) {
+    return Math.abs(value) < EPSILON ? 0 : value;
   }
 
   @Override
@@ -253,6 +292,6 @@ public void reset() {
 
   @Override
   public boolean removable() {
-    return false;
+    return true;
   }
 }

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCorrelationAccumulator.java

@@ -36,6 +36,7 @@ public class TableCorrelationAccumulator implements TableAccumulator {
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(TableCorrelationAccumulator.class);
   private static final int INTERMEDIATE_SIZE = Long.BYTES + 5 * Double.BYTES;
+  private static final double EPSILON = 1e-12;
   private final TSDataType xDataType;
   private final TSDataType yDataType;
   private final CorrelationAccumulator.CorrelationType correlationType;
@@ -126,7 +127,42 @@ private void update(double x, double y) {
 
   @Override
   public void removeInput(Column[] arguments) {
-    throw new UnsupportedOperationException("Remove not implemented for Correlation Accumulator");
+    checkArgument(arguments.length == 2, "Input of Correlation should be 2");
+    if (count == 0 || arguments[0].isNull(0) || arguments[1].isNull(0)) {
+      return;
+    }
+
+    double otherX = getDoubleValue(arguments[0], 0, xDataType);
+    double otherY = getDoubleValue(arguments[1], 0, yDataType);
+
+    if (count == 1) {
+      reset();
+      return;
+    }
+
+    long totalCount = count;
+    long newCount = totalCount - 1;
+
+    double newMeanX = (totalCount * meanX - otherX) / newCount;
+    double newMeanY = (totalCount * meanY - otherY) / newCount;
+
+    double deltaX = otherX - newMeanX;
+    double deltaY = otherY - newMeanY;
+    double correction = (double) newCount / totalCount;
+
+    c2 = c2 - deltaX * deltaY * correction;
+    m2X = m2X - deltaX * deltaX * correction;
+    m2Y = m2Y - deltaY * deltaY * correction;
+    meanX = newMeanX;
+    meanY = newMeanY;
+    c2 = normalizeZero(c2);
+    m2X = normalizeZero(m2X);
+    m2Y = normalizeZero(m2Y);
+    count = newCount;
+  }
+
+  private double normalizeZero(double value) {
+    return Math.abs(value) < EPSILON ? 0 : value;
   }
 
   @Override
@@ -247,6 +283,6 @@ public void reset() {
 
   @Override
   public boolean removable() {
-    return false;
+    return true;
   }
 }

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCovarianceAccumulator.java

@@ -36,6 +36,7 @@ public class TableCovarianceAccumulator implements TableAccumulator {
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(TableCovarianceAccumulator.class);
   private static final int INTERMEDIATE_SIZE = Long.BYTES + 3 * Double.BYTES;
+  private static final double EPSILON = 1e-12;
 
   private final TSDataType xDataType;
   private final TSDataType yDataType;
@@ -123,7 +124,38 @@ private void update(double x, double y) {
 
   @Override
   public void removeInput(Column[] arguments) {
-    throw new UnsupportedOperationException("Remove not implemented for Covariance Accumulator");
+    checkArgument(arguments.length == 2, "Input of Covariance should be 2");
+    if (count == 0 || arguments[0].isNull(0) || arguments[1].isNull(0)) {
+      return;
+    }
+
+    double otherX = getDoubleValue(arguments[0], 0, xDataType);
+    double otherY = getDoubleValue(arguments[1], 0, yDataType);
+
+    if (count == 1) {
+      reset();
+      return;
+    }
+
+    long totalCount = count;
+    long newCount = totalCount - 1;
+
+    double newMeanX = (totalCount * meanX - otherX) / newCount;
+    double newMeanY = (totalCount * meanY - otherY) / newCount;
+
+    double deltaX = otherX - newMeanX;
+    double deltaY = otherY - newMeanY;
+    double correction = (double) newCount / totalCount;
+
+    c2 = c2 - deltaX * deltaY * correction;
+    meanX = newMeanX;
+    meanY = newMeanY;
+    c2 = normalizeZero(c2);
+    count = newCount;
+  }
+
+  private double normalizeZero(double value) {
+    return Math.abs(value) < EPSILON ? 0 : value;
   }
 
   @Override
@@ -235,6 +267,6 @@ public void reset() {
 
   @Override
   public boolean removable() {
-    return false;
+    return true;
   }
 }

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableRegressionAccumulator.java

@@ -36,6 +36,7 @@ public class TableRegressionAccumulator implements TableAccumulator {
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(TableRegressionAccumulator.class);
   private static final int INTERMEDIATE_SIZE = Long.BYTES + 4 * Double.BYTES;
+  private static final double EPSILON = 1e-12;
 
   private final TSDataType yDataType;
   private final TSDataType xDataType;
@@ -220,7 +221,40 @@ public void evaluateFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public void removeInput(Column[] arguments) {
-    throw new UnsupportedOperationException();
+    checkArgument(arguments.length == 2, "Input of Regression should be 2");
+    if (count == 0 || arguments[0].isNull(0) || arguments[1].isNull(0)) {
+      return;
+    }
+
+    double otherY = getDoubleValue(arguments[0], 0, yDataType);
+    double otherX = getDoubleValue(arguments[1], 0, xDataType);
+
+    if (count == 1) {
+      reset();
+      return;
+    }
+
+    long totalCount = count;
+    long newCount = totalCount - 1;
+
+    double newMeanX = (totalCount * meanX - otherX) / newCount;
+    double newMeanY = (totalCount * meanY - otherY) / newCount;
+
+    double deltaX = otherX - newMeanX;
+    double deltaY = otherY - newMeanY;
+    double correction = (double) newCount / totalCount;
+
+    c2 = c2 - deltaX * deltaY * correction;
+    m2X = m2X - deltaX * deltaX * correction;
+    meanX = newMeanX;
+    meanY = newMeanY;
+    c2 = normalizeZero(c2);
+    m2X = normalizeZero(m2X);
+    count = newCount;
+  }
+
+  private double normalizeZero(double value) {
+    return Math.abs(value) < EPSILON ? 0 : value;
   }
 
   @Override
@@ -244,6 +278,6 @@ public void reset() {
 
   @Override
   public boolean removable() {
-    return false;
+    return true;
   }
 }