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 <[email protected]> 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); + } + } +}
