You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/10/23 09:43:45 UTC
(iotdb) 01/02: add variance
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch add_more_aggs
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7d3285c639d4f22e8531b830a32a8735a200f919
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Oct 23 17:40:02 2024 +0800
add variance
---
.../relational/aggregation/AccumulatorFactory.java | 15 ++
.../aggregation/TableVarianceAccumulator.java | 187 +++++++++++++++++++++
2 files changed, 202 insertions(+)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
index cb59c7b2ac3..5003883eb45 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
+import org.apache.iotdb.db.queryengine.execution.aggregation.VarianceAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAvgAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountAccumulator;
@@ -181,6 +182,20 @@ public class AccumulatorFactory {
return new ExtremeAccumulator(inputDataTypes.get(0));
case MODE:
return new TableModeAccumulator(inputDataTypes.get(0));
+ case STDDEV:
+ case STDDEV_SAMP:
+ return new TableVarianceAccumulator(
+ inputDataTypes.get(0), VarianceAccumulator.VarianceType.STDDEV_SAMP);
+ case STDDEV_POP:
+ return new TableVarianceAccumulator(
+ inputDataTypes.get(0), VarianceAccumulator.VarianceType.STDDEV_POP);
+ case VARIANCE:
+ case VAR_SAMP:
+ return new TableVarianceAccumulator(
+ inputDataTypes.get(0), VarianceAccumulator.VarianceType.VAR_SAMP);
+ case VAR_POP:
+ return new TableVarianceAccumulator(
+ inputDataTypes.get(0), VarianceAccumulator.VarianceType.VAR_POP);
default:
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
new file mode 100644
index 00000000000..4cfb95c016f
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation;
+
+import org.apache.iotdb.db.queryengine.execution.aggregation.VarianceAccumulator;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BytesUtils;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+public class TableVarianceAccumulator implements TableAccumulator {
+
+ private final TSDataType seriesDataType;
+
+ private final VarianceAccumulator.VarianceType varianceType;
+
+ private long count;
+ private double mean;
+ private double m2;
+
+ public TableVarianceAccumulator(
+ TSDataType seriesDataType, VarianceAccumulator.VarianceType varianceType) {
+ this.seriesDataType = seriesDataType;
+ this.varianceType = varianceType;
+ }
+
+ @Override
+ public long getEstimatedSize() {
+ return 0;
+ }
+
+ @Override
+ public TableAccumulator copy() {
+ return null;
+ }
+
+ @Override
+ public void addInput(Column[] arguments) {
+ switch (seriesDataType) {
+ case INT32:
+ addIntInput(arguments[0]);
+ return;
+ case INT64:
+ // addLongInput(arguments[0]);
+ return;
+ case FLOAT:
+ // addFloatInput(arguments[0]);
+ return;
+ case DOUBLE:
+ // addDoubleInput(arguments[0]);
+ return;
+ case TEXT:
+ case BLOB:
+ case BOOLEAN:
+ case DATE:
+ case STRING:
+ case TIMESTAMP:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation variance : %s", seriesDataType));
+ }
+ }
+
+ @Override
+ public void addIntermediate(Column argument) {
+ for (int i = 0; i < argument.getPositionCount(); i++) {
+ if (argument.isNull(i)) {
+ continue;
+ }
+
+ byte[] bytes = argument.getBinary(i).getValues();
+ long intermediateCount = BytesUtils.bytesToLong(bytes, Long.BYTES);
+ double intermediateMean = BytesUtils.bytesToDouble(bytes, Long.BYTES);
+ double intermediateM2 = BytesUtils.bytesToDouble(bytes, (Long.BYTES + Double.BYTES));
+
+ long newCount = count + intermediateCount;
+ double newMean = ((intermediateCount * intermediateMean) + (count * mean)) / newCount;
+ double delta = intermediateMean - mean;
+
+ m2 = m2 + intermediateM2 + delta * delta * intermediateCount * count / newCount;
+ count = newCount;
+ mean = newMean;
+ }
+ }
+
+ @Override
+ public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+ if (count == 0) {
+ columnBuilder.appendNull();
+ } else {
+ byte[] bytes = new byte[24];
+ BytesUtils.longToBytes(count, bytes, 0);
+ BytesUtils.doubleToBytes(mean, bytes, Long.BYTES);
+ BytesUtils.doubleToBytes(m2, bytes, Long.BYTES + Double.BYTES);
+ columnBuilder.writeBinary(new Binary(bytes));
+ }
+ }
+
+ @Override
+ public void evaluateFinal(ColumnBuilder columnBuilder) {
+ switch (varianceType) {
+ case STDDEV_POP:
+ if (count == 0) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(Math.sqrt(m2 / count));
+ }
+ break;
+ case STDDEV_SAMP:
+ if (count < 2) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(Math.sqrt(m2 / (count - 1)));
+ }
+ break;
+ case VAR_POP:
+ if (count == 0) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(m2 / count);
+ }
+ break;
+ case VAR_SAMP:
+ if (count < 2) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(m2 / (count - 1));
+ }
+ break;
+ default:
+ throw new EnumConstantNotPresentException(
+ VarianceAccumulator.VarianceType.class, varianceType.name());
+ }
+ }
+
+ @Override
+ public boolean hasFinalResult() {
+ return false;
+ }
+
+ @Override
+ public void addStatistics(Statistics[] statistics) {
+ throw new UnsupportedOperationException(getClass().getName());
+ }
+
+ @Override
+ public void reset() {
+ count = 0;
+ mean = 0.0;
+ m2 = 0.0;
+ }
+
+ private void addIntInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (column.isNull(i)) {
+ continue;
+ }
+
+ int value = column.getInt(i);
+ count++;
+ double delta = value - mean;
+ mean += delta / count;
+ m2 += delta * (value - mean);
+ }
+ }
+}