Implement removeIntermediate method for CentralMoment, Correlation, Covariance, and Regression accumulators to handle state removal

Author: CoollZzz

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CentralMomentAccumulator.java

@@ -198,7 +198,66 @@ public void outputFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public void removeIntermediate(Column[] input) {
-    throw new UnsupportedOperationException();
+    checkArgument(input.length == 1, "Input of CentralMoment should be 1");
+    if (input[0].isNull(0)) {
+      return;
+    }
+
+    byte[] bytes = input[0].getBinary(0).getValues();
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+    long nB = buffer.getLong();
+    if (nB == 0) {
+      return;
+    }
+    checkArgument(count >= nB, "CentralMoment state count is smaller than removed state count");
+
+    if (count == nB) {
+      reset();
+      return;
+    }
+
+    double meanB = buffer.getDouble();
+    double m2B = buffer.getDouble();
+    double m3B = buffer.getDouble();
+    double m4B = buffer.getDouble();
+
+    long nTotal = count;
+    long nA = nTotal - nB;
+
+    double meanA = ((double) nTotal * mean - (double) nB * meanB) / nA;
+
+    double delta = meanB - meanA;
+    double delta2 = delta * delta;
+    double delta3 = delta * delta2;
+    double delta4 = delta2 * delta2;
+
+    double m2A = m2 - m2B - delta2 * nA * nB / nTotal;
+    double m3A =
+        m3
+            - m3B
+            - delta3 * nA * nB * (nA - nB) / ((double) nTotal * nTotal)
+            - 3.0 * delta * (nA * m2B - nB * m2A) / nTotal;
+
+    double m4A =
+        m4
+            - m4B
+            - delta4
+                * nA
+                * nB
+                * ((double) nA * nA - (double) nA * nB + (double) nB * nB)
+                / ((double) nTotal * nTotal * nTotal)
+            - 6.0
+                * delta2
+                * ((double) nA * nA * m2B + (double) nB * nB * m2A)
+                / ((double) nTotal * nTotal)
+            - 4.0 * delta * (nA * m3B - nB * m3A) / nTotal;
+
+    count = nA;
+    mean = meanA;
+    m2 = m2A;
+    m3 = m3A;
+    m4 = m4A;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CorrelationAccumulator.java

@@ -187,8 +187,49 @@ public void outputFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public void removeIntermediate(Column[] input) {
+    checkArgument(input.length == 1, "Input of Correlation should be 1");
+    if (input[0].isNull(0)) {
+      return;
+    }
+
+    byte[] bytes = input[0].getBinary(0).getValues();
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+    long otherCount = buffer.getLong();
+    if (otherCount == 0) {
+      return;
+    }
+    checkArgument(
+        count >= otherCount, "Correlation state count is smaller than removed state count");
+
+    if (count == otherCount) {
+      reset();
+      return;
+    }
 
-    throw new UnsupportedOperationException("Remove not implemented for Correlation");
+    double otherMeanX = buffer.getDouble();
+    double otherMeanY = buffer.getDouble();
+    double otherM2X = buffer.getDouble();
+    double otherM2Y = buffer.getDouble();
+    double otherC2 = buffer.getDouble();
+
+    long totalCount = count;
+    long newCount = totalCount - otherCount;
+
+    double newMeanX = (totalCount * meanX - otherCount * otherMeanX) / newCount;
+    double newMeanY = (totalCount * meanY - otherCount * otherMeanY) / newCount;
+
+    double deltaX = otherMeanX - newMeanX;
+    double deltaY = otherMeanY - newMeanY;
+    double correction = ((double) newCount * otherCount) / totalCount;
+
+    c2 = c2 - otherC2 - deltaX * deltaY * correction;
+    m2X = m2X - otherM2X - deltaX * deltaX * correction;
+    m2Y = m2Y - otherM2Y - deltaY * deltaY * correction;
+
+    meanX = newMeanX;
+    meanY = newMeanY;
+    count = newCount;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/CovarianceAccumulator.java

@@ -174,7 +174,44 @@ public void outputFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public void removeIntermediate(Column[] input) {
-    throw new UnsupportedOperationException("Remove not implemented for Covariance");
+    checkArgument(input.length == 1, "Input of Covariance should be 1");
+    if (input[0].isNull(0)) {
+      return;
+    }
+
+    byte[] bytes = input[0].getBinary(0).getValues();
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+    long otherCount = buffer.getLong();
+    if (otherCount == 0) {
+      return;
+    }
+    checkArgument(
+        count >= otherCount, "Covariance state count is smaller than removed state count");
+
+    if (count == otherCount) {
+      reset();
+      return;
+    }
+
+    double otherMeanX = buffer.getDouble();
+    double otherMeanY = buffer.getDouble();
+    double otherC2 = buffer.getDouble();
+
+    long totalCount = count;
+    long newCount = totalCount - otherCount;
+
+    double newMeanX = ((double) totalCount * meanX - (double) otherCount * otherMeanX) / newCount;
+    double newMeanY = ((double) totalCount * meanY - (double) otherCount * otherMeanY) / newCount;
+
+    double deltaX = otherMeanX - newMeanX;
+    double deltaY = otherMeanY - newMeanY;
+    double correction = ((double) newCount * otherCount) / totalCount;
+
+    c2 = c2 - otherC2 - deltaX * deltaY * correction;
+    meanX = newMeanX;
+    meanY = newMeanY;
+    count = newCount;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/RegressionAccumulator.java

@@ -183,7 +183,47 @@ public void outputFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public void removeIntermediate(Column[] input) {
-    throw new UnsupportedOperationException();
+    checkArgument(input.length == 1, "Input of Regression should be 1");
+    if (input[0].isNull(0)) {
+      return;
+    }
+
+    byte[] bytes = input[0].getBinary(0).getValues();
+    ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+    long otherCount = buffer.getLong();
+    if (otherCount == 0) {
+      return;
+    }
+    checkArgument(
+        count >= otherCount, "Regression state count is smaller than removed state count");
+
+    if (count == otherCount) {
+      reset();
+      return;
+    }
+
+    double otherMeanX = buffer.getDouble();
+    double otherMeanY = buffer.getDouble();
+    double otherM2X = buffer.getDouble();
+    double otherC2 = buffer.getDouble();
+
+    long totalCount = count;
+    long newCount = totalCount - otherCount;
+
+    double newMeanX = ((double) totalCount * meanX - (double) otherCount * otherMeanX) / newCount;
+    double newMeanY = ((double) totalCount * meanY - (double) otherCount * otherMeanY) / newCount;
+
+    double deltaX = otherMeanX - newMeanX;
+    double deltaY = otherMeanY - newMeanY;
+    double correction = ((double) newCount * otherCount) / totalCount;
+
+    c2 = c2 - otherC2 - deltaX * deltaY * correction;
+    m2X = m2X - otherM2X - deltaX * deltaX * correction;
+
+    meanX = newMeanX;
+    meanY = newMeanY;
+    count = newCount;
   }
 
   @Override