This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e363388966a Add mode and variance aggregation support for table model
e363388966a is described below
commit e363388966ad47f0388ef768ab884dd70797600f
Author: Beyyes <[email protected]>
AuthorDate: Thu Oct 24 09:47:43 2024 +0800
Add mode and variance aggregation support for table model
---
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 36 ++
.../relational/aggregation/AccumulatorFactory.java | 17 +
.../aggregation/AggregationOperator.java | 2 +-
.../aggregation/TableModeAccumulator.java | 480 +++++++++++++++++++++
.../aggregation/TableVarianceAccumulator.java | 231 ++++++++++
.../source/relational/aggregation/Utils.java | 9 +-
.../plan/planner/TableOperatorGenerator.java | 70 +--
7 files changed, 813 insertions(+), 32 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
index 69f52276de6..ec659eddb42 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
@@ -1197,6 +1197,42 @@ public class IoTDBMultiIDsWithAttributesTableIT {
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}
+ @Test
+ public void modeTest() {
+ expectedHeader = buildHeaders(15);
+ sql =
+ "select
mode(time),mode(device),mode(level),mode(attr1),mode(attr2),mode(num),mode(bignum),mode(floatnum),mode(date),mode(str),mode(bool),mode(date),mode(ts),mode(stringv),mode(doublenum)
from table0 where device='d2' and level='l4' and time=80";
+ retArray =
+ new String[] {
+
"1970-01-01T00:00:00.080Z,d2,l4,null,null,9,2147483646,43.12,null,apple,false,null,2024-09-20T06:15:35.000Z,test-string2,6666.7,",
+ };
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ expectedHeader = buildHeaders(10);
+ sql =
+ "select
mode(device),mode(level),mode(attr1),mode(attr2),mode(date),mode(bool),mode(date),mode(ts),mode(stringv),mode(doublenum)
from table0 where device='d2' and level='l1'";
+ retArray =
+ new String[] {
+
"d2,l1,d,c,null,false,null,2024-08-01T06:15:35.000Z,test-string3,null,",
+ };
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+ }
+
+ @Test
+ public void varianceTest() {
+ expectedHeader = buildHeaders(18);
+ sql =
+ "select \n"
+ +
"round(variance(num),1),round(var_pop(num),1),round(var_samp(num),1),round(stddev(num),1),round(stddev_pop(num),1),round(stddev_samp(num),1),\n"
+ +
"round(variance(floatnum),1),round(var_pop(floatnum),1),round(var_samp(floatnum),1),round(stddev(floatnum),1),round(stddev_pop(floatnum),1),round(stddev_samp(floatnum),1),\n"
+ +
"round(variance(doublenum),1),round(var_pop(doublenum),1),round(var_samp(doublenum),1),round(stddev(doublenum),1),round(stddev_pop(doublenum),1),round(stddev_samp(doublenum),1)
from table0 where device='d2' and level='l4'";
+ retArray =
+ new String[] {
+
"16.0,10.7,16.0,4.0,3.3,4.0,50.0,33.3,50.0,7.1,5.8,7.1,null,0.0,null,null,0.0,null,",
+ };
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+ }
+
// ==================================================================
// ============================ Join Test ===========================
// ==================================================================
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 8875bcfacaa..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;
@@ -179,6 +180,22 @@ public class AccumulatorFactory {
return new TableMinByAccumulator(inputDataTypes.get(0),
inputDataTypes.get(1));
case EXTREME:
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/AggregationOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
index 3da8ba5c84f..81abe6a62f1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AggregationOperator.java
@@ -151,7 +151,7 @@ public class AggregationOperator implements ProcessOperator
{
public long ramBytesUsed() {
return INSTANCE_SIZE
+ MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(child)
- +
aggregators.stream().mapToLong(TableAggregator::getEstimatedSize).count()
+ +
aggregators.stream().mapToLong(TableAggregator::getEstimatedSize).sum()
+
MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ resultBuilder.getRetainedSizeInBytes();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
new file mode 100644
index 00000000000..3f90db0a1bd
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
@@ -0,0 +1,480 @@
+/*
+ * 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.conf.IoTDBDescriptor;
+
+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.utils.RamUsageEstimator;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.UNSUPPORTED_TYPE_MESSAGE;
+import static
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.serializeBinaryValue;
+
+public class TableModeAccumulator implements TableAccumulator {
+
+ private final int MAP_SIZE_THRESHOLD =
+ IoTDBDescriptor.getInstance().getConfig().getModeMapSizeThreshold();
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(TableModeAccumulator.class);
+ private final TSDataType seriesDataType;
+
+ private Map<Boolean, Long> booleanCountMap;
+ private Map<Integer, Long> intCountMap;
+ private Map<Float, Long> floatCountMap;
+ private Map<Long, Long> longCountMap;
+ private Map<Double, Long> doubleCountMap;
+ private Map<Binary, Long> binaryCountMap;
+
+ public TableModeAccumulator(TSDataType seriesDataType) {
+ this.seriesDataType = seriesDataType;
+ switch (seriesDataType) {
+ case BOOLEAN:
+ booleanCountMap = new HashMap<>();
+ break;
+ case INT32:
+ case DATE:
+ intCountMap = new HashMap<>();
+ break;
+ case FLOAT:
+ floatCountMap = new HashMap<>();
+ break;
+ case INT64:
+ case TIMESTAMP:
+ longCountMap = new HashMap<>();
+ break;
+ case DOUBLE:
+ doubleCountMap = new HashMap<>();
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ binaryCountMap = new HashMap<>();
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format(UNSUPPORTED_TYPE_MESSAGE, seriesDataType));
+ }
+ }
+
+ @Override
+ public long getEstimatedSize() {
+ return INSTANCE_SIZE;
+ }
+
+ @Override
+ public TableAccumulator copy() {
+ return new TableModeAccumulator(seriesDataType);
+ }
+
+ @Override
+ public void addInput(Column[] arguments) {
+ switch (seriesDataType) {
+ case BOOLEAN:
+ addBooleanInput(arguments[0]);
+ break;
+ case INT32:
+ case DATE:
+ addIntInput(arguments[0]);
+ break;
+ case FLOAT:
+ addFloatInput(arguments[0]);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ addLongInput(arguments[0]);
+ break;
+ case DOUBLE:
+ addDoubleInput(arguments[0]);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ addBinaryInput(arguments[0]);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format(UNSUPPORTED_TYPE_MESSAGE, 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();
+ deserializeAndMergeCountMap(bytes);
+ }
+ }
+
+ @Override
+ public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+ columnBuilder.writeBinary(new Binary(serializeCountMap()));
+ }
+
+ @Override
+ public void evaluateFinal(ColumnBuilder columnBuilder) {
+ switch (seriesDataType) {
+ case BOOLEAN:
+ if (booleanCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Boolean> maxKey =
+ booleanCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeBoolean);
+ }
+ break;
+ case INT32:
+ case DATE:
+ if (intCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Integer> maxKey =
+ intCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeInt);
+ }
+ break;
+ case FLOAT:
+ if (floatCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Float> maxKey =
+ floatCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeFloat);
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ if (longCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Long> maxKey =
+ longCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeLong);
+ }
+ break;
+ case DOUBLE:
+ if (doubleCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Double> maxKey =
+ doubleCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeDouble);
+ }
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ if (binaryCountMap.isEmpty()) {
+ columnBuilder.appendNull();
+ } else {
+ Optional<Binary> maxKey =
+ binaryCountMap.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ maxKey.ifPresent(columnBuilder::writeBinary);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format(UNSUPPORTED_TYPE_MESSAGE, seriesDataType));
+ }
+ }
+
+ @Override
+ public boolean hasFinalResult() {
+ return false;
+ }
+
+ @Override
+ public void addStatistics(Statistics[] statistics) {
+ throw new UnsupportedOperationException(getClass().getName());
+ }
+
+ @Override
+ public void reset() {
+ if (booleanCountMap != null) {
+ booleanCountMap.clear();
+ }
+ if (intCountMap != null) {
+ intCountMap.clear();
+ }
+ if (floatCountMap != null) {
+ floatCountMap.clear();
+ }
+ if (longCountMap != null) {
+ longCountMap.clear();
+ }
+ if (doubleCountMap != null) {
+ doubleCountMap.clear();
+ }
+ if (binaryCountMap != null) {
+ binaryCountMap.clear();
+ }
+ }
+
+ private byte[] serializeCountMap() {
+ byte[] bytes;
+ int offset = 0;
+
+ switch (seriesDataType) {
+ case BOOLEAN:
+ bytes = new byte[4 + (1 + 8) * booleanCountMap.size()];
+ BytesUtils.intToBytes(booleanCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Boolean, Long> entry : booleanCountMap.entrySet()) {
+ BytesUtils.boolToBytes(entry.getKey(), bytes, offset);
+ offset += 1;
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ case INT32:
+ case DATE:
+ bytes = new byte[4 + (4 + 8) * intCountMap.size()];
+ BytesUtils.intToBytes(intCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Integer, Long> entry : intCountMap.entrySet()) {
+ BytesUtils.intToBytes(entry.getKey(), bytes, offset);
+ offset += 4;
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ case FLOAT:
+ bytes = new byte[4 + (4 + 8) * floatCountMap.size()];
+ BytesUtils.intToBytes(floatCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Float, Long> entry : floatCountMap.entrySet()) {
+ BytesUtils.floatToBytes(entry.getKey(), bytes, offset);
+ offset += 4;
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ bytes = new byte[4 + (8 + 8) * longCountMap.size()];
+ BytesUtils.intToBytes(longCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Long, Long> entry : longCountMap.entrySet()) {
+ BytesUtils.longToBytes(entry.getKey(), bytes, offset);
+ offset += 8;
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ case DOUBLE:
+ bytes = new byte[4 + (8 + 8) * doubleCountMap.size()];
+ BytesUtils.intToBytes(doubleCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Double, Long> entry : doubleCountMap.entrySet()) {
+ BytesUtils.doubleToBytes(entry.getKey(), bytes, offset);
+ offset += 8;
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ bytes =
+ new byte
+ [4
+ + (8 + 4) * binaryCountMap.size()
+ + binaryCountMap.keySet().stream()
+ .mapToInt(key -> key.getValues().length)
+ .sum()];
+ BytesUtils.intToBytes(binaryCountMap.size(), bytes, offset);
+ offset += 4;
+ for (Map.Entry<Binary, Long> entry : binaryCountMap.entrySet()) {
+ Binary binary = entry.getKey();
+ serializeBinaryValue(binary, bytes, offset);
+ offset += (4 + binary.getLength());
+ BytesUtils.longToBytes(entry.getValue(), bytes, offset);
+ offset += 8;
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format(UNSUPPORTED_TYPE_MESSAGE, seriesDataType));
+ }
+
+ return bytes;
+ }
+
+ private void deserializeAndMergeCountMap(byte[] bytes) {
+ int offset = 0;
+ int size = BytesUtils.bytesToInt(bytes, offset);
+ offset += 4;
+
+ switch (seriesDataType) {
+ case BOOLEAN:
+ for (int i = 0; i < size; i++) {
+ boolean key = BytesUtils.bytesToBool(bytes, offset);
+ offset += 1;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ booleanCountMap.compute(key, (k, v) -> v == null ? count : v +
count);
+ }
+ break;
+ case INT32:
+ case DATE:
+ for (int i = 0; i < size; i++) {
+ int key = BytesUtils.bytesToInt(bytes, offset);
+ offset += 4;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ intCountMap.compute(key, (k, v) -> v == null ? count : v + count);
+ }
+ break;
+ case FLOAT:
+ for (int i = 0; i < size; i++) {
+ float key = BytesUtils.bytesToFloat(bytes, offset);
+ offset += 4;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ floatCountMap.compute(key, (k, v) -> v == null ? count : v + count);
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ for (int i = 0; i < size; i++) {
+ long key = BytesUtils.bytesToLong(bytes, offset);
+ offset += 8;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ longCountMap.compute(key, (k, v) -> v == null ? count : v + count);
+ }
+ break;
+ case DOUBLE:
+ for (int i = 0; i < size; i++) {
+ double key = BytesUtils.bytesToDouble(bytes, offset);
+ offset += 8;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ doubleCountMap.compute(key, (k, v) -> v == null ? count : v + count);
+ }
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ for (int i = 0; i < size; i++) {
+ int length = BytesUtils.bytesToInt(bytes, offset);
+ offset += 4;
+ Binary binaryVal = new Binary(BytesUtils.subBytes(bytes, offset,
length));
+ offset += length;
+ long count = BytesUtils.bytesToLongFromOffset(bytes, 8, offset);
+ offset += 8;
+ binaryCountMap.compute(binaryVal, (k, v) -> v == null ? count : v +
count);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ String.format(UNSUPPORTED_TYPE_MESSAGE, seriesDataType));
+ }
+ }
+
+ private void addBooleanInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ booleanCountMap.compute(column.getBoolean(i), (k, v) -> v == null ? 1
: v + 1);
+ if (booleanCountMap.size() > MAP_SIZE_THRESHOLD) {
+ checkMapSize(booleanCountMap.size());
+ }
+ }
+ }
+ }
+
+ private void addIntInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ intCountMap.compute(column.getInt(i), (k, v) -> v == null ? 1 : v + 1);
+ checkMapSize(intCountMap.size());
+ }
+ }
+ }
+
+ private void addFloatInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ floatCountMap.compute(column.getFloat(i), (k, v) -> v == null ? 1 : v
+ 1);
+ checkMapSize(floatCountMap.size());
+ }
+ }
+ }
+
+ private void addLongInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ longCountMap.compute(column.getLong(i), (k, v) -> v == null ? 1 : v +
1);
+ checkMapSize(longCountMap.size());
+ }
+ }
+ }
+
+ private void addDoubleInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ doubleCountMap.compute(column.getDouble(i), (k, v) -> v == null ? 1 :
v + 1);
+ checkMapSize(doubleCountMap.size());
+ }
+ }
+ }
+
+ private void addBinaryInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (!column.isNull(i)) {
+ binaryCountMap.compute(column.getBinary(i), (k, v) -> v == null ? 1 :
v + 1);
+ checkMapSize(binaryCountMap.size());
+ }
+ }
+ }
+
+ private void checkMapSize(int size) {
+ if (size > MAP_SIZE_THRESHOLD) {
+ throw new RuntimeException(
+ String.format(
+ "distinct values has exceeded the threshold %s when calculate
Mode",
+ MAP_SIZE_THRESHOLD));
+ }
+ }
+}
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..0eb60b15991
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
@@ -0,0 +1,231 @@
+/*
+ * 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.utils.RamUsageEstimator;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+public class TableVarianceAccumulator implements TableAccumulator {
+
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(TableVarianceAccumulator.class);
+ 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 INSTANCE_SIZE;
+ }
+
+ @Override
+ public TableAccumulator copy() {
+ return new TableVarianceAccumulator(seriesDataType, varianceType);
+ }
+
+ @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);
+ }
+ }
+
+ private void addLongInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (column.isNull(i)) {
+ continue;
+ }
+
+ long value = column.getLong(i);
+ count++;
+ double delta = value - mean;
+ mean += delta / count;
+ m2 += delta * (value - mean);
+ }
+ }
+
+ private void addFloatInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (column.isNull(i)) {
+ continue;
+ }
+
+ float value = column.getFloat(i);
+ count++;
+ double delta = value - mean;
+ mean += delta / count;
+ m2 += delta * (value - mean);
+ }
+ }
+
+ private void addDoubleInput(Column column) {
+ for (int i = 0; i < column.getPositionCount(); i++) {
+ if (column.isNull(i)) {
+ continue;
+ }
+
+ double value = column.getDouble(i);
+ count++;
+ double delta = value - mean;
+ mean += delta / count;
+ m2 += delta * (value - mean);
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Utils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Utils.java
index 85c79fd8ea0..99252b9606d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Utils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/Utils.java
@@ -20,6 +20,7 @@
package
org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.TsPrimitiveType;
import org.apache.tsfile.write.UnSupportedDataTypeException;
@@ -29,7 +30,7 @@ import static org.apache.tsfile.enums.TSDataType.STRING;
import static org.apache.tsfile.enums.TSDataType.TEXT;
public class Utils {
- public static final String UNSUPPORTED_TYPE_MESSAGE = "Unsupported data type
in serialize : %s";
+ public static final String UNSUPPORTED_TYPE_MESSAGE = "Unsupported data type
: %s";
private Utils() {}
@@ -70,6 +71,12 @@ public class Utils {
}
}
+ public static void serializeBinaryValue(Binary binary, byte[] valueBytes,
int offset) {
+ BytesUtils.intToBytes(binary.getValues().length, valueBytes, offset);
+ offset += 4;
+ System.arraycopy(binary.getValues(), 0, valueBytes, offset,
binary.getValues().length);
+ }
+
public static byte[] serializeTimeValue(
TSDataType seriesDataType, long time, TsPrimitiveType value) {
byte[] valueBytes = new byte[8 + calcTypeSize(seriesDataType, value)];
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index fbd146ab82d..1d02608b317 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -188,14 +188,16 @@ import static
org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator
import static
org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.getAggregationTypeByFuncName;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder.ASC_NULLS_LAST;
import static
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager.getTSDataType;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.AVG;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.EXTREME;
import static org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_AGGREGATION;
import static
org.apache.iotdb.db.utils.constant.SqlConstant.FIRST_BY_AGGREGATION;
import static org.apache.iotdb.db.utils.constant.SqlConstant.LAST_AGGREGATION;
import static
org.apache.iotdb.db.utils.constant.SqlConstant.LAST_BY_AGGREGATION;
import static org.apache.iotdb.db.utils.constant.SqlConstant.MAX;
-import static org.apache.iotdb.db.utils.constant.SqlConstant.MAX_BY;
import static org.apache.iotdb.db.utils.constant.SqlConstant.MIN;
-import static org.apache.iotdb.db.utils.constant.SqlConstant.MIN_BY;
+import static org.apache.iotdb.db.utils.constant.SqlConstant.SUM;
import static org.apache.tsfile.read.common.type.TimestampType.TIMESTAMP;
/** This Visitor is responsible for transferring Table PlanNode Tree to Table
Operator Tree. */
@@ -1686,37 +1688,45 @@ public class TableOperatorGenerator extends
PlanVisitor<Operator, LocalExecution
Symbol argument = Symbol.from(aggregation.getArguments().get(0));
Type argumentType = node.getAssignments().get(argument).getType();
- if (MAX_BY.equals(funcName) || MIN_BY.equals(funcName)) {
- canUseStatistic = false;
- } else if (MAX.equals(funcName) || MIN.equals(funcName)) {
- if (BlobType.BLOB.equals(argumentType)
- || BinaryType.TEXT.equals(argumentType)
- || BooleanType.BOOLEAN.equals(argumentType)) {
- canUseStatistic = false;
- }
- } else if (FIRST_AGGREGATION.equals(funcName)
- || LAST_AGGREGATION.equals(funcName)
- || LAST_BY_AGGREGATION.equals(funcName)
- || FIRST_BY_AGGREGATION.equals(funcName)) {
- if (FIRST_AGGREGATION.equals(funcName) ||
FIRST_BY_AGGREGATION.equals(funcName)) {
- ascendingCount++;
- } else {
- descendingCount++;
- }
-
- // first/last/first_by/last_by aggregation with BLOB type can not use
statistics
+ switch (funcName) {
+ case COUNT:
+ case AVG:
+ case SUM:
+ case EXTREME:
+ break;
+ case MAX:
+ case MIN:
+ if (BlobType.BLOB.equals(argumentType)
+ || BinaryType.TEXT.equals(argumentType)
+ || BooleanType.BOOLEAN.equals(argumentType)) {
+ canUseStatistic = false;
+ }
+ break;
+ case FIRST_AGGREGATION:
+ case LAST_AGGREGATION:
+ case LAST_BY_AGGREGATION:
+ case FIRST_BY_AGGREGATION:
+ if (FIRST_AGGREGATION.equals(funcName) ||
FIRST_BY_AGGREGATION.equals(funcName)) {
+ ascendingCount++;
+ } else {
+ descendingCount++;
+ }
- if (BlobType.BLOB.equals(argumentType)) {
- canUseStatistic = false;
- continue;
- }
+ // first/last/first_by/last_by aggregation with BLOB type can not
use statistics
+ if (BlobType.BLOB.equals(argumentType)) {
+ canUseStatistic = false;
+ break;
+ }
- // only last_by(time, x) or last_by(x,time) can use statistic
- if ((LAST_BY_AGGREGATION.equals(funcName) ||
FIRST_BY_AGGREGATION.equals(funcName))
- && !isTimeColumn(aggregation.getArguments().get(0))
- && !isTimeColumn(aggregation.getArguments().get(1))) {
+ // only last_by(time, x) or last_by(x,time) can use statistic
+ if ((LAST_BY_AGGREGATION.equals(funcName) ||
FIRST_BY_AGGREGATION.equals(funcName))
+ && !isTimeColumn(aggregation.getArguments().get(0))
+ && !isTimeColumn(aggregation.getArguments().get(1))) {
+ canUseStatistic = false;
+ }
+ break;
+ default:
canUseStatistic = false;
- }
}
}
