You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/10/17 12:37:41 UTC
(iotdb) branch master updated: Implement hash grouped aggregation in TableModel
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 e854bd79011 Implement hash grouped aggregation in TableModel
e854bd79011 is described below
commit e854bd79011107ee5cf09e2c6f64260c9a714ea5
Author: Weihao Li <60...@users.noreply.github.com>
AuthorDate: Thu Oct 17 20:37:35 2024 +0800
Implement hash grouped aggregation in TableModel
---
.../relational/aggregation/AccumulatorFactory.java | 35 ++
.../aggregation/grouped/GroupedAccumulator.java | 34 ++
.../aggregation/grouped/GroupedAggregator.java | 91 ++++
.../aggregation/grouped/GroupedAvgAccumulator.java | 186 +++++++
.../grouped/GroupedCountAccumulator.java | 73 +++
.../grouped/HashAggregationOperator.java | 211 ++++++++
.../aggregation/grouped/UpdateMemory.java | 26 +
.../aggregation/grouped/array/BigArrays.java | 56 ++
.../aggregation/grouped/array/BooleanBigArray.java | 156 ++++++
.../aggregation/grouped/array/ByteBigArray.java | 152 ++++++
.../aggregation/grouped/array/DoubleBigArray.java | 165 ++++++
.../aggregation/grouped/array/IntBigArray.java | 182 +++++++
.../aggregation/grouped/array/IntBigArrays.java | 188 +++++++
.../aggregation/grouped/array/IntComparator.java | 27 +
.../aggregation/grouped/array/LongBigArray.java | 186 +++++++
.../aggregation/grouped/array/ObjectBigArray.java | 173 ++++++
.../aggregation/grouped/array/ShortBigArray.java | 176 ++++++
.../grouped/builder/HashAggregationBuilder.java | 31 ++
.../builder/InMemoryHashAggregationBuilder.java | 227 ++++++++
.../grouped/hash/CombineHashFunction.java | 31 ++
.../aggregation/grouped/hash/FlatGroupByHash.java | 204 +++++++
.../aggregation/grouped/hash/FlatHash.java | 599 +++++++++++++++++++++
.../aggregation/grouped/hash/FlatHashStrategy.java | 49 ++
.../aggregation/grouped/hash/GroupByHash.java | 53 ++
.../aggregation/grouped/hash/HashStrategy.java | 225 ++++++++
.../grouped/hash/VariableWidthData.java | 191 +++++++
.../aggregation/grouped/hash/XxHash64.java | 276 ++++++++++
.../plan/planner/TableOperatorGenerator.java | 95 +++-
.../planner/node/AggregationTableScanNode.java | 5 +-
.../plan/relational/utils/TypeUtil.java | 397 ++++++++++++++
30 files changed, 4496 insertions(+), 4 deletions(-)
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 4cf810289bf..7ad725699d7 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,9 @@
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.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;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.ConstantOperand;
@@ -50,6 +53,38 @@ public class AccumulatorFactory {
}
}
+ public static GroupedAccumulator createGroupedAccumulator(
+ String functionName,
+ TAggregationType aggregationType,
+ List<TSDataType> inputDataTypes,
+ List<Expression> inputExpressions,
+ Map<String, String> inputAttributes,
+ boolean ascending) {
+ if (aggregationType == TAggregationType.UDAF) {
+ // If UDAF accumulator receives raw input, it needs to check input's attribute
+ throw new UnsupportedOperationException();
+ } else {
+ return createBuiltinGroupedAccumulator(
+ aggregationType, inputDataTypes, inputExpressions, inputAttributes, ascending);
+ }
+ }
+
+ private static GroupedAccumulator createBuiltinGroupedAccumulator(
+ TAggregationType aggregationType,
+ List<TSDataType> inputDataTypes,
+ List<Expression> inputExpressions,
+ Map<String, String> inputAttributes,
+ boolean ascending) {
+ switch (aggregationType) {
+ case COUNT:
+ return new GroupedCountAccumulator();
+ case AVG:
+ return new GroupedAvgAccumulator(inputDataTypes.get(0));
+ default:
+ throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
+ }
+ }
+
public static TableAccumulator createBuiltinAccumulator(
TAggregationType aggregationType,
List<TSDataType> inputDataTypes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java
new file mode 100644
index 00000000000..f417af0dfcd
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAccumulator.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed 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.grouped;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+public interface GroupedAccumulator {
+
+ long getEstimatedSize();
+
+ void setGroupCount(long groupCount);
+
+ void addInput(int[] groupIds, Column[] arguments);
+
+ void addIntermediate(int[] groupIds, Column argument);
+
+ void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder);
+
+ void evaluateFinal(int groupId, ColumnBuilder columnBuilder);
+
+ void prepareFinal();
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java
new file mode 100644
index 00000000000..4e73193aab9
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAggregator.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed 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.grouped;
+
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
+
+import com.google.common.primitives.Ints;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+
+import java.util.List;
+import java.util.OptionalInt;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public class GroupedAggregator {
+ private final GroupedAccumulator accumulator;
+ private final AggregationNode.Step step;
+ private final TSDataType outputType;
+ private final int[] inputChannels;
+ private final OptionalInt maskChannel;
+
+ public GroupedAggregator(
+ GroupedAccumulator accumulator,
+ AggregationNode.Step step,
+ TSDataType outputType,
+ List<Integer> inputChannels,
+ OptionalInt maskChannel) {
+ this.accumulator = requireNonNull(accumulator, "accumulator is null");
+ this.step = requireNonNull(step, "step is null");
+ this.outputType = requireNonNull(outputType, "intermediateType is null");
+ this.inputChannels = Ints.toArray(requireNonNull(inputChannels, "inputChannels is null"));
+ this.maskChannel = requireNonNull(maskChannel, "maskChannel is null");
+ checkArgument(
+ step.isInputRaw() || inputChannels.size() == 1,
+ "expected 1 input channel for intermediate aggregation");
+ }
+
+ public TSDataType getType() {
+ return outputType;
+ }
+
+ public void processBlock(int groupCount, int[] groupIds, TsBlock block) {
+ accumulator.setGroupCount(groupCount);
+ Column[] arguments = block.getColumns(inputChannels);
+
+ // process count(*)
+ if (arguments.length == 0) {
+ arguments =
+ new Column[] {new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, block.getPositionCount())};
+ }
+
+ if (step.isInputRaw()) {
+ accumulator.addInput(groupIds, arguments);
+ } else {
+ accumulator.addIntermediate(groupIds, arguments[0]);
+ }
+ }
+
+ public void evaluate(int groupId, ColumnBuilder columnBuilder) {
+ if (step.isOutputPartial()) {
+ accumulator.evaluateIntermediate(groupId, columnBuilder);
+ } else {
+ accumulator.evaluateFinal(groupId, columnBuilder);
+ }
+ }
+
+ public void prepareFinal() {
+ accumulator.prepareFinal();
+ }
+
+ public long getEstimatedSize() {
+ return accumulator.getEstimatedSize();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAvgAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAvgAccumulator.java
new file mode 100644
index 00000000000..c975d177273
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedAvgAccumulator.java
@@ -0,0 +1,186 @@
+/*
+ * 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.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.DoubleBigArray;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.LongBigArray;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BytesUtils;
+import org.apache.tsfile.utils.RamUsageEstimator;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+public class GroupedAvgAccumulator implements GroupedAccumulator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(GroupedAvgAccumulator.class);
+ private final TSDataType argumentDataType;
+ private final LongBigArray countValues = new LongBigArray();
+ private final DoubleBigArray sumValues = new DoubleBigArray();
+
+ public GroupedAvgAccumulator(TSDataType argumentDataType) {
+ this.argumentDataType = argumentDataType;
+ }
+
+ @Override
+ public long getEstimatedSize() {
+ return INSTANCE_SIZE + countValues.sizeOf() + sumValues.sizeOf();
+ }
+
+ @Override
+ public void setGroupCount(long groupCount) {
+ countValues.ensureCapacity(groupCount);
+ sumValues.ensureCapacity(groupCount);
+ }
+
+ @Override
+ public void addInput(int[] groupIds, Column[] arguments) {
+ checkArgument(arguments.length == 1, "argument of Avg should be one column");
+ switch (argumentDataType) {
+ case INT32:
+ addIntInput(groupIds, arguments[0]);
+ return;
+ case INT64:
+ addLongInput(groupIds, arguments[0]);
+ return;
+ case FLOAT:
+ addFloatInput(groupIds, arguments[0]);
+ return;
+ case DOUBLE:
+ addDoubleInput(groupIds, arguments[0]);
+ return;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ case BOOLEAN:
+ case DATE:
+ case TIMESTAMP:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type in aggregation AVG : %s", argumentDataType));
+ }
+ }
+
+ @Override
+ public void addIntermediate(int[] groupIds, Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn,
+ "intermediate input and output of Avg should be BinaryColumn");
+ if (argument.isNull(0)) {
+ return;
+ }
+
+ for (int i = 0; i < groupIds.length; i++) {
+ deserialize(groupIds[i], argument.getBinary(i).getValues());
+ }
+ }
+
+ @Override
+ public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output of Avg should be BinaryColumn");
+ if (countValues.get(groupId) == 0) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeBinary(new Binary(serializeState(groupId)));
+ }
+ }
+
+ private void deserialize(int groupId, byte[] bytes) {
+ countValues.add(groupId, BytesUtils.bytesToLong(bytes, Long.BYTES));
+ sumValues.add(groupId, BytesUtils.bytesToDouble(bytes, Long.BYTES));
+ }
+
+ private byte[] serializeState(int groupId) {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ try {
+ dataOutputStream.writeLong(countValues.get(groupId));
+ dataOutputStream.writeDouble(sumValues.get(groupId));
+ } catch (IOException e) {
+ throw new UnsupportedOperationException(
+ "Failed to serialize intermediate result for AvgAccumulator.", e);
+ }
+ return byteArrayOutputStream.toByteArray();
+ }
+
+ private void addIntInput(int[] groupIds, Column column) {
+ int count = column.getPositionCount();
+ for (int i = 0; i < count; i++) {
+ if (!column.isNull(i)) {
+ countValues.increment(groupIds[i]);
+ sumValues.add(groupIds[i], column.getInt(i));
+ }
+ }
+ }
+
+ private void addLongInput(int[] groupIds, Column column) {
+ int count = column.getPositionCount();
+ for (int i = 0; i < count; i++) {
+ if (!column.isNull(i)) {
+ countValues.increment(groupIds[i]);
+ sumValues.add(groupIds[i], column.getLong(i));
+ }
+ }
+ }
+
+ private void addFloatInput(int[] groupIds, Column column) {
+ int count = column.getPositionCount();
+ for (int i = 0; i < count; i++) {
+ if (!column.isNull(i)) {
+ countValues.increment(groupIds[i]);
+ sumValues.add(groupIds[i], column.getFloat(i));
+ }
+ }
+ }
+
+ private void addDoubleInput(int[] groupIds, Column column) {
+ int count = column.getPositionCount();
+ for (int i = 0; i < count; i++) {
+ if (!column.isNull(i)) {
+ countValues.increment(groupIds[i]);
+ sumValues.add(groupIds[i], column.getDouble(i));
+ }
+ }
+ }
+
+ @Override
+ public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) {
+ long countValue = countValues.get(groupId);
+ if (countValue == 0) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.writeDouble(sumValues.get(groupId) / countValue);
+ }
+ }
+
+ @Override
+ public void prepareFinal() {}
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAccumulator.java
new file mode 100644
index 00000000000..2d75b7b824b
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedCountAccumulator.java
@@ -0,0 +1,73 @@
+/*
+ * 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.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.LongBigArray;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+public class GroupedCountAccumulator implements GroupedAccumulator {
+ private final LongBigArray countValues = new LongBigArray(0L);
+
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(GroupedCountAccumulator.class);
+
+ @Override
+ public long getEstimatedSize() {
+ return INSTANCE_SIZE + countValues.sizeOf();
+ }
+
+ @Override
+ public void setGroupCount(long groupCount) {
+ countValues.ensureCapacity(groupCount);
+ }
+
+ @Override
+ public void addInput(int[] groupIds, Column[] arguments) {
+ for (int i = 0; i < groupIds.length; i++) {
+ if (!arguments[0].isNull(i)) {
+ countValues.increment(groupIds[i]);
+ }
+ }
+ }
+
+ @Override
+ public void addIntermediate(int[] groupIds, Column argument) {
+ for (int i = 0; i < groupIds.length; i++) {
+ if (!argument.isNull(i)) {
+ countValues.add(groupIds[i], argument.getLong(i));
+ }
+ }
+ }
+
+ @Override
+ public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ columnBuilder.writeLong(countValues.get(groupId));
+ }
+
+ @Override
+ public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) {
+ columnBuilder.writeLong(countValues.get(groupId));
+ }
+
+ @Override
+ public void prepareFinal() {}
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java
new file mode 100644
index 00000000000..ee395a64128
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/HashAggregationOperator.java
@@ -0,0 +1,211 @@
+/*
+ * 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.grouped;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.builder.HashAggregationBuilder;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.builder.InMemoryHashAggregationBuilder;
+import org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.List;
+import java.util.Optional;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory.NOOP;
+
+public class HashAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(HashAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<Type> groupByTypes;
+ private final List<Integer> groupByChannels;
+
+ private final List<GroupedAggregator> aggregators;
+ private final AggregationNode.Step step;
+
+ private final int expectedGroups;
+
+ private final long maxPartialMemory;
+
+ private final boolean spillEnabled;
+ private final long unspillMemoryLimit;
+
+ private HashAggregationBuilder aggregationBuilder;
+
+ protected MemoryReservationManager memoryReservationManager;
+
+ private boolean finished = false;
+
+ public HashAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ List<GroupedAggregator> aggregators,
+ AggregationNode.Step step,
+ int expectedGroups,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unspillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByTypes = ImmutableList.copyOf(groupByTypes);
+ this.groupByChannels = ImmutableList.copyOf(groupByChannels);
+ this.aggregators = aggregators;
+ this.step = step;
+ this.expectedGroups = expectedGroups;
+ this.maxPartialMemory = maxPartialMemory;
+ this.spillEnabled = spillEnabled;
+ this.unspillMemoryLimit = unspillMemoryLimit;
+ this.memoryReservationManager =
+ operatorContext
+ .getDriverContext()
+ .getFragmentInstanceContext()
+ .getMemoryReservationContext();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished;
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (aggregationBuilder == null) {
+ if (spillEnabled) {
+ throw new UnsupportedOperationException();
+ } else {
+ aggregationBuilder =
+ new InMemoryHashAggregationBuilder(
+ aggregators,
+ step,
+ expectedGroups,
+ groupByTypes,
+ groupByChannels,
+ Optional.empty(),
+ operatorContext,
+ maxPartialMemory,
+ NOOP);
+ }
+ } else {
+ checkState(!aggregationBuilder.isFull(), "Aggregation buffer is full");
+ }
+
+ // Each call only calculate at most once, no need to check time slice.
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ aggregationBuilder.processBlock(block);
+ aggregationBuilder.updateMemory();
+ return null;
+ } else {
+ // evaluate output
+ resultTsBlock = getOutput();
+ return checkTsBlockSizeAndGetResult();
+ }
+ }
+
+ private TsBlock getOutput() {
+ // only flush if we are finishing or the aggregation builder is full
+ if (aggregationBuilder == null) {
+ return null;
+ }
+
+ TsBlock result = aggregationBuilder.buildResult();
+
+ closeAggregationBuilder();
+ finished = true;
+ return result;
+ }
+
+ private void closeAggregationBuilder() {
+ // outputPages = null;
+ if (aggregationBuilder != null) {
+ aggregationBuilder.close();
+ // aggregationBuilder.close() will release all memory reserved in memory accounting.
+ // The reference must be set to null afterwards to avoid unaccounted memory.
+ aggregationBuilder = null;
+ }
+ // memoryContext.setBytes(0);
+ }
+
+ @Override
+ public boolean isFinished() throws Exception {
+ return finished;
+ }
+
+ @Override
+ public void close() throws Exception {
+ child.close();
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public long calculateMaxPeekMemory() {
+ return Math.max(
+ child.calculateMaxPeekMemoryWithCounter(),
+ calculateRetainedSizeAfterCallingNext() + calculateMaxReturnSize());
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return maxReturnSize;
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ return child.calculateMaxReturnSize() + child.calculateRetainedSizeAfterCallingNext();
+ }
+
+ @Override
+ public long ramBytesUsed() {
+ return INSTANCE_SIZE
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(child)
+ + (aggregationBuilder == null ? 0 : aggregationBuilder.getEstimatedSize())
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/UpdateMemory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/UpdateMemory.java
new file mode 100644
index 00000000000..a1d9e6f2e21
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/UpdateMemory.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed 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.grouped;
+
+public interface UpdateMemory {
+ UpdateMemory NOOP = () -> true;
+
+ /**
+ * Update memory reservation.
+ *
+ * @return false if the caller should yield because requested memory reservation cannot be
+ * satisfied
+ */
+ boolean update();
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BigArrays.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BigArrays.java
new file mode 100644
index 00000000000..2cbb40c8a40
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BigArrays.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed 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.grouped.array;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class BigArrays {
+ private BigArrays() {}
+
+ /** Initial number of segments to support in array. */
+ static final int INITIAL_SEGMENTS = 1024;
+
+ /**
+ * The shift used to compute the segment associated with an index (equivalently, the logarithm of
+ * the segment size).
+ */
+ static final int SEGMENT_SHIFT = 10;
+
+ /** Size of a single segment of a BigArray. */
+ public static final int SEGMENT_SIZE = 1 << SEGMENT_SHIFT;
+
+ /** The mask used to compute the offset associated to an index. */
+ static final int SEGMENT_MASK = SEGMENT_SIZE - 1;
+
+ /**
+ * Computes the segment associated with a given index.
+ *
+ * @param index an index into a big array.
+ * @return the associated segment.
+ */
+ @SuppressWarnings("NumericCastThatLosesPrecision")
+ static int segment(long index) {
+ return (int) (index >>> SEGMENT_SHIFT);
+ }
+
+ /**
+ * Computes the offset associated with a given index.
+ *
+ * @param index an index into a big array.
+ * @return the associated offset (in the associated {@linkplain #segment(long) segment}).
+ */
+ static int offset(long index) {
+ return (int) (index & SEGMENT_MASK);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BooleanBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BooleanBigArray.java
new file mode 100644
index 00000000000..a85d50c1f91
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BooleanBigArray.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfBooleanArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class BooleanBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(BooleanBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfBooleanArray(SEGMENT_SIZE);
+
+ private final boolean initialValue;
+
+ private boolean[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public BooleanBigArray() {
+ this(false);
+ }
+
+ public BooleanBigArray(boolean initialValue) {
+ this.initialValue = initialValue;
+ array = new boolean[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public boolean get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, boolean value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ public long getCapacity() {
+ return capacity;
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(boolean value) {
+ for (boolean[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, BooleanBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ boolean[] newSegment = new boolean[SEGMENT_SIZE];
+ if (initialValue) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ByteBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ByteBigArray.java
new file mode 100644
index 00000000000..fe9ce4bb2c7
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ByteBigArray.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfByteArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class ByteBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(ByteBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfByteArray(SEGMENT_SIZE);
+
+ private final byte initialValue;
+
+ private byte[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public ByteBigArray() {
+ this((byte) 0);
+ }
+
+ public ByteBigArray(byte initialValue) {
+ this.initialValue = initialValue;
+ array = new byte[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public byte get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, byte value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(byte value) {
+ for (byte[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, ByteBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ byte[] newSegment = new byte[SEGMENT_SIZE];
+ if (initialValue != 0) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/DoubleBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/DoubleBigArray.java
new file mode 100644
index 00000000000..644d81e03a5
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/DoubleBigArray.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfDoubleArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class DoubleBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(DoubleBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfDoubleArray(SEGMENT_SIZE);
+
+ private final double initialValue;
+
+ private double[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public DoubleBigArray() {
+ this(0.0);
+ }
+
+ /**
+ * Creates a new big array containing one initial segment filled with the specified default value
+ */
+ public DoubleBigArray(double initialValue) {
+ this.initialValue = initialValue;
+ array = new double[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public double get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, double value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Adds the specified value to the specified element of this big array.
+ *
+ * @param index a position in this big array.
+ * @param value the value
+ */
+ public void add(long index, double value) {
+ array[segment(index)][offset(index)] += value;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(double value) {
+ for (double[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, DoubleBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ double[] newSegment = new double[SEGMENT_SIZE];
+ if (initialValue != 0.0) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArray.java
new file mode 100644
index 00000000000..1ca45e897c4
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArray.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfIntArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class IntBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(IntBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfIntArray(SEGMENT_SIZE);
+
+ private final int initialValue;
+
+ private int[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public IntBigArray() {
+ this(0);
+ }
+
+ /**
+ * Creates a new big array containing one initial segment filled with the specified default value
+ */
+ public IntBigArray(int initialValue) {
+ this.initialValue = initialValue;
+ array = new int[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ public int[][] getSegments() {
+ return array;
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public int get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, int value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Increments the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void increment(long index) {
+ array[segment(index)][offset(index)]++;
+ }
+
+ /**
+ * Adds the specified value to the specified element of this big array.
+ *
+ * @param index a position in this big array.
+ * @param value the value
+ */
+ public void add(long index, int value) {
+ array[segment(index)][offset(index)] += value;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(int value) {
+ for (int[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, IntBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ int[] newSegment = new int[SEGMENT_SIZE];
+ if (initialValue != 0) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+
+ public void sort(int from, int to, IntComparator comparator) {
+ IntBigArrays.quickSort(array, from, to, comparator);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArrays.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArrays.java
new file mode 100644
index 00000000000..919cc08c42d
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntBigArrays.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed 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.grouped.array;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SHIFT;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class IntBigArrays {
+ private static final int SEGMENT_MASK = SEGMENT_SIZE - 1;
+ private static final int SMALL = 7;
+ private static final int MEDIUM = 40;
+
+ private IntBigArrays() {}
+
+ /**
+ * Computes the segment associated with a given index.
+ *
+ * @param index an index into a big array.
+ * @return the associated segment.
+ */
+ public static int segment(final long index) {
+ return (int) (index >>> SEGMENT_SHIFT);
+ }
+
+ /**
+ * Computes the displacement associated with a given index.
+ *
+ * @param index an index into a big array.
+ * @return the associated displacement (in the associated {@linkplain #segment(long) segment}).
+ */
+ public static int displacement(final long index) {
+ return (int) (index & SEGMENT_MASK);
+ }
+
+ /**
+ * Returns the element of the given big array of specified index.
+ *
+ * @param array a big array.
+ * @param index a position in the big array.
+ * @return the element of the big array at the specified position.
+ */
+ public static int get(final int[][] array, final long index) {
+ return array[segment(index)][displacement(index)];
+ }
+
+ /**
+ * Sets the element of the given big array of specified index.
+ *
+ * @param array a big array.
+ * @param index a position in the big array.
+ */
+ public static void set(final int[][] array, final long index, int value) {
+ array[segment(index)][displacement(index)] = value;
+ }
+
+ /**
+ * Swaps the element of the given big array of specified indices.
+ *
+ * @param array a big array.
+ * @param first a position in the big array.
+ * @param second a position in the big array.
+ */
+ public static void swap(final int[][] array, final long first, final long second) {
+ final int t = array[segment(first)][displacement(first)];
+ array[segment(first)][displacement(first)] = array[segment(second)][displacement(second)];
+ array[segment(second)][displacement(second)] = t;
+ }
+
+ /**
+ * Sorts the specified range of elements according to the order induced by the specified
+ * comparator using quicksort.
+ *
+ * <p>
+ *
+ * <p>The sorting algorithm is a tuned quicksort adapted from Jon L. Bentley and M. Douglas
+ * McIlroy, “Engineering a Sort Function”, <i>Software: Practice and Experience</i>,
+ * 23(11), pages 1249−1265, 1993.
+ *
+ * @param x the big array to be sorted.
+ * @param from the index of the first element (inclusive) to be sorted.
+ * @param to the index of the last element (exclusive) to be sorted.
+ * @param comp the comparator to determine the sorting order.
+ */
+ @SuppressWarnings("checkstyle:InnerAssignment")
+ public static void quickSort(
+ final int[][] x, final long from, final long to, final IntComparator comp) {
+ final long len = to - from;
+ // Selection sort on smallest arrays
+ if (len < SMALL) {
+ selectionSort(x, from, to, comp);
+ return;
+ }
+ // Choose a partition element, v
+ long m = from + len / 2; // Small arrays, middle element
+ if (len > SMALL) {
+ long l = from;
+ long n = to - 1;
+ if (len > MEDIUM) { // Big arrays, pseudomedian of 9
+ long s = len / 8;
+ l = med3(x, l, l + s, l + 2 * s, comp);
+ m = med3(x, m - s, m, m + s, comp);
+ n = med3(x, n - 2 * s, n - s, n, comp);
+ }
+ m = med3(x, l, m, n, comp); // Mid-size, med of 3
+ }
+ final int v = get(x, m);
+ // Establish Invariant: v* (<v)* (>v)* v*
+ long a = from;
+ long b = a;
+ long c = to - 1;
+ long d = c;
+ while (true) {
+ int comparison;
+ while (b <= c && (comparison = comp.compare(get(x, b), v)) <= 0) {
+ if (comparison == 0) {
+ swap(x, a++, b);
+ }
+ b++;
+ }
+ while (c >= b && (comparison = comp.compare(get(x, c), v)) >= 0) {
+ if (comparison == 0) {
+ swap(x, c, d--);
+ }
+ c--;
+ }
+ if (b > c) {
+ break;
+ }
+ swap(x, b++, c--);
+ }
+ // Swap partition elements back to middle
+ long s;
+ long n = to;
+ s = Math.min(a - from, b - a);
+ vecSwap(x, from, b - s, s);
+ s = Math.min(d - c, n - d - 1);
+ vecSwap(x, b, n - s, s);
+ // Recursively sort non-partition-elements
+ if ((s = b - a) > 1) {
+ quickSort(x, from, from + s, comp);
+ }
+ if ((s = d - c) > 1) {
+ quickSort(x, n - s, n, comp);
+ }
+ }
+
+ private static void vecSwap(final int[][] x, long a, long b, final long n) {
+ for (int i = 0; i < n; i++, a++, b++) {
+ swap(x, a, b);
+ }
+ }
+
+ private static long med3(
+ final int[][] x, final long a, final long b, final long c, IntComparator comp) {
+ int ab = comp.compare(get(x, a), get(x, b));
+ int ac = comp.compare(get(x, a), get(x, c));
+ int bc = comp.compare(get(x, b), get(x, c));
+ return (ab < 0 ? (bc < 0 ? b : ac < 0 ? c : a) : (bc > 0 ? b : ac > 0 ? c : a));
+ }
+
+ private static void selectionSort(
+ final int[][] a, final long from, final long to, final IntComparator comp) {
+ for (long i = from; i < to - 1; i++) {
+ long m = i;
+ for (long j = i + 1; j < to; j++) {
+ if (comp.compare(IntBigArrays.get(a, j), IntBigArrays.get(a, m)) < 0) {
+ m = j;
+ }
+ }
+ if (m != i) {
+ swap(a, i, m);
+ }
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntComparator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntComparator.java
new file mode 100644
index 00000000000..fb0259a4a50
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/IntComparator.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed 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.grouped.array;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public interface IntComparator {
+ /**
+ * Compares the given primitive types.
+ *
+ * @return A positive integer, zero, or a negative integer if the first argument is greater than,
+ * equal to, or smaller than, respectively, the second one.
+ * @see java.util.Comparator
+ */
+ int compare(int k1, int k2);
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/LongBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/LongBigArray.java
new file mode 100644
index 00000000000..3e4a1f74f72
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/LongBigArray.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfLongArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class LongBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(LongBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfLongArray(SEGMENT_SIZE);
+
+ private final long initialValue;
+
+ private long[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public LongBigArray() {
+ this(0L);
+ }
+
+ /**
+ * Creates a new big array containing one initial segment filled with the specified default value
+ */
+ public LongBigArray(long initialValue) {
+ this.initialValue = initialValue;
+ array = new long[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public long get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ public long[] getSegment(long index) {
+ return array[segment(index)];
+ }
+
+ public int getOffset(long index) {
+ return offset(index);
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, long value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Increments the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void increment(long index) {
+ array[segment(index)][offset(index)]++;
+ }
+
+ /**
+ * Adds the specified value to the specified element of this big array.
+ *
+ * @param index a position in this big array.
+ * @param value the value
+ */
+ public void add(long index, long value) {
+ array[segment(index)][offset(index)] += value;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ public long getCapacity() {
+ return capacity;
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, LongBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(long value) {
+ for (long[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ long[] newSegment = new long[SEGMENT_SIZE];
+ if (initialValue != 0) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ObjectBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ObjectBigArray.java
new file mode 100644
index 00000000000..14690d24ca8
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ObjectBigArray.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed 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.grouped.array;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfObjectArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class ObjectBigArray<T> {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(ObjectBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfObjectArray(SEGMENT_SIZE);
+
+ private final Object initialValue;
+
+ private Object[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public ObjectBigArray() {
+ this(null);
+ }
+
+ public ObjectBigArray(Object initialValue) {
+ this.initialValue = initialValue;
+ array = new Object[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the current available capacity in this array */
+ public long getCapacity() {
+ return capacity;
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ @SuppressWarnings("unchecked")
+ public T get(long index) {
+ return (T) array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, T value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Replaces the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return true if the previous value was not null
+ */
+ public boolean replace(long index, T value) {
+ Object[] segment = array[segment(index)];
+
+ boolean existed = segment[offset(index)] != null;
+ segment[offset(index)] = value;
+
+ return existed;
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(T value) {
+ for (Object[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, ObjectBigArray<T> destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ Object[] newSegment = new Object[SEGMENT_SIZE];
+ if (initialValue != null) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ShortBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ShortBigArray.java
new file mode 100644
index 00000000000..275f654488c
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/ShortBigArray.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed 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.grouped.array;
+
+import com.google.common.primitives.Shorts;
+
+import java.util.Arrays;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.INITIAL_SEGMENTS;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.SEGMENT_SIZE;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.offset;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.BigArrays.segment;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfShortArray;
+
+// Note: this code was forked from fastutil (http://fastutil.di.unimi.it/)
+// Copyright (C) 2010-2013 Sebastiano Vigna
+public final class ShortBigArray {
+ private static final long INSTANCE_SIZE = shallowSizeOfInstance(ShortBigArray.class);
+ private static final long SIZE_OF_SEGMENT = sizeOfShortArray(SEGMENT_SIZE);
+
+ private final short initialValue;
+
+ private short[][] array;
+ private long capacity;
+ private int segments;
+
+ /** Creates a new big array containing one initial segment */
+ public ShortBigArray() {
+ this((short) 0);
+ }
+
+ /**
+ * Creates a new big array containing one initial segment filled with the specified default value
+ */
+ public ShortBigArray(short initialValue) {
+ this.initialValue = initialValue;
+ array = new short[INITIAL_SEGMENTS][];
+ allocateNewSegment();
+ }
+
+ /** Returns the size of this big array in bytes. */
+ public long sizeOf() {
+ return INSTANCE_SIZE + shallowSizeOf(array) + (segments * SIZE_OF_SEGMENT);
+ }
+
+ /**
+ * Returns the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ * @return the element of this big array at the specified position.
+ */
+ public short get(long index) {
+ return array[segment(index)][offset(index)];
+ }
+
+ /**
+ * Sets the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void set(long index, short value) {
+ array[segment(index)][offset(index)] = value;
+ }
+
+ /**
+ * Increments the element of this big array at specified index.
+ *
+ * @param index a position in this big array.
+ */
+ public void increment(long index) {
+ array[segment(index)][offset(index)]++;
+ }
+
+ /**
+ * Adds the specified value to the specified element of this big array.
+ *
+ * @param index a position in this big array.
+ * @param value the value
+ */
+ public void add(long index, long value) {
+ array[segment(index)][offset(index)] += Shorts.checkedCast(value);
+ }
+
+ /**
+ * Ensures this big array is at least the specified length. If the array is smaller, segments are
+ * added until the array is larger then the specified length.
+ */
+ public void ensureCapacity(long length) {
+ if (capacity > length) {
+ return;
+ }
+
+ grow(length);
+ }
+
+ /** Fills the entire big array with the specified value. */
+ public void fill(short value) {
+ for (short[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ Arrays.fill(segment, value);
+ }
+ }
+
+ /**
+ * Copies this array, beginning at the specified sourceIndex, to the specified destinationIndex of
+ * the destination array. A subsequence of this array's components are copied to the destination
+ * array referenced by {@code destination}. The number of components copied is equal to the {@code
+ * length} argument. The components at positions {@code sourceIndex} through {@code
+ * sourceIndex+length-1} in this array are copied into positions {@code destinationIndex} through
+ * {@code destinationIndex+length-1}, respectively, of the destination array.
+ */
+ public void copyTo(
+ long sourceIndex, ShortBigArray destination, long destinationIndex, long length) {
+ while (length > 0) {
+ int startSegment = segment(sourceIndex);
+ int startOffset = offset(sourceIndex);
+ int destinationStartSegment = segment(destinationIndex);
+ int destinationStartOffset = offset(destinationIndex);
+
+ int copyLength = Math.min(SEGMENT_SIZE - startOffset, SEGMENT_SIZE - destinationStartOffset);
+ copyLength =
+ Math.min(copyLength, length > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) length);
+
+ System.arraycopy(
+ array[startSegment],
+ startOffset,
+ destination.array[destinationStartSegment],
+ destinationStartOffset,
+ copyLength);
+
+ sourceIndex += copyLength;
+ destinationIndex += copyLength;
+ length -= copyLength;
+ }
+ }
+
+ private void grow(long length) {
+ // how many segments are required to get to the length?
+ int requiredSegments = segment(length) + 1;
+
+ // grow base array if necessary
+ if (array.length < requiredSegments) {
+ array = Arrays.copyOf(array, requiredSegments);
+ }
+
+ // add new segments
+ while (segments < requiredSegments) {
+ allocateNewSegment();
+ }
+ }
+
+ private void allocateNewSegment() {
+ short[] newSegment = new short[SEGMENT_SIZE];
+ if (initialValue != 0) {
+ Arrays.fill(newSegment, initialValue);
+ }
+ array[segments] = newSegment;
+ capacity += SEGMENT_SIZE;
+ segments++;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/HashAggregationBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/HashAggregationBuilder.java
new file mode 100644
index 00000000000..0a420e68de1
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/HashAggregationBuilder.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed 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.grouped.builder;
+
+import org.apache.tsfile.read.common.block.TsBlock;
+
+public interface HashAggregationBuilder extends AutoCloseable {
+ void processBlock(TsBlock block);
+
+ TsBlock buildResult();
+
+ long getEstimatedSize();
+
+ boolean isFull();
+
+ void updateMemory();
+
+ @Override
+ void close();
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/InMemoryHashAggregationBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/InMemoryHashAggregationBuilder.java
new file mode 100644
index 00000000000..7ddcc9adabb
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/builder/InMemoryHashAggregationBuilder.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed 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.grouped.builder;
+
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAggregator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.GroupByHash;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
+import org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.primitives.Ints;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.GroupByHash.createGroupByHash;
+
+public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
+ private final int[] groupByChannels;
+ private final GroupByHash groupByHash;
+ private final List<Type> groupByOutputTypes;
+ private final List<GroupedAggregator> groupedAggregators;
+ private final boolean partial;
+ private final OptionalLong maxPartialMemory;
+ private final UpdateMemory updateMemory;
+
+ private boolean full;
+
+ public InMemoryHashAggregationBuilder(
+ List<GroupedAggregator> groupedAggregators,
+ AggregationNode.Step step,
+ int expectedGroups,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ Optional<Integer> hashChannel,
+ OperatorContext operatorContext,
+ long maxPartialMemory,
+ UpdateMemory updateMemory) {
+ this(
+ groupedAggregators,
+ step,
+ expectedGroups,
+ groupByTypes,
+ groupByChannels,
+ hashChannel,
+ operatorContext,
+ maxPartialMemory,
+ Optional.empty(),
+ updateMemory);
+ }
+
+ public InMemoryHashAggregationBuilder(
+ List<GroupedAggregator> groupedAggregators,
+ AggregationNode.Step step,
+ int expectedGroups,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ Optional<Integer> hashChannel,
+ OperatorContext operatorContext,
+ long maxPartialMemory,
+ Optional<Integer> unspillIntermediateChannelOffset,
+ UpdateMemory updateMemory) {
+ this.groupedAggregators = groupedAggregators;
+ this.groupByOutputTypes = ImmutableList.copyOf(groupByTypes);
+ this.groupByChannels = Ints.toArray(groupByChannels);
+
+ this.groupByHash =
+ createGroupByHash(groupByTypes, hashChannel.isPresent(), expectedGroups, updateMemory);
+ this.partial = step.isOutputPartial();
+ this.maxPartialMemory = OptionalLong.of(maxPartialMemory);
+ this.updateMemory = updateMemory;
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void processBlock(TsBlock block) {
+ if (groupedAggregators.isEmpty()) {
+ groupByHash.addPage(block.getColumns(groupByChannels));
+ }
+ int[] groupByIdBlock = groupByHash.getGroupIds(block.getColumns(groupByChannels));
+
+ int groupCount = groupByHash.getGroupCount();
+ for (GroupedAggregator groupedAggregator : groupedAggregators) {
+ groupedAggregator.processBlock(groupCount, groupByIdBlock, block);
+ }
+ }
+
+ @Override
+ public void updateMemory() {
+ // updateMemory.update();
+ }
+
+ @Override
+ public boolean isFull() {
+ return full;
+ }
+
+ public long getEstimatedSize() {
+ long sizeInMemory = groupByHash.getEstimatedSize();
+ for (GroupedAggregator groupedAggregator : groupedAggregators) {
+ sizeInMemory += groupedAggregator.getEstimatedSize();
+ }
+
+ updateIsFull(sizeInMemory);
+ return sizeInMemory;
+ }
+
+ private void updateIsFull(long sizeInMemory) {
+ if (!partial || !maxPartialMemory.isPresent()) {
+ return;
+ }
+
+ full = sizeInMemory > maxPartialMemory.getAsLong();
+ }
+
+ /**
+ * building hash sorted results requires memory for sorting group IDs. This method returns size of
+ * that memory requirement.
+ */
+ public long getGroupIdsSortingSize() {
+ return getGroupCount() * Integer.BYTES;
+ }
+
+ public void setSpillOutput() {
+ for (GroupedAggregator groupedAggregator : groupedAggregators) {
+ // groupedAggregator.setSpillOutput();
+ }
+ }
+
+ public int getKeyChannels() {
+ return groupByChannels.length;
+ }
+
+ public long getGroupCount() {
+ return groupByHash.getGroupCount();
+ }
+
+ @Override
+ public TsBlock buildResult() {
+ // used for order by in Agg
+ /*for (GroupedAggregator groupedAggregator : groupedAggregators) {
+ groupedAggregator.prepareFinal();
+ }*/
+ return buildResult(consecutiveGroupIds());
+ }
+
+ public List<Type> buildSpillTypes() {
+ ArrayList<Type> types = new ArrayList<>(groupByOutputTypes);
+ for (GroupedAggregator groupedAggregator : groupedAggregators) {
+ // types.add(groupedAggregator.getSpillType());
+ }
+ return types;
+ }
+
+ public int getCapacity() {
+ return groupByHash.getCapacity();
+ }
+
+ private TsBlock buildResult(Iterator<Integer> groupIds) {
+ TsBlockBuilder pageBuilder = new TsBlockBuilder(buildTypes());
+
+ pageBuilder.reset();
+
+ while (!pageBuilder.isFull() && groupIds.hasNext()) {
+ int groupId = groupIds.next();
+
+ groupByHash.appendValuesTo(groupId, pageBuilder);
+
+ pageBuilder.declarePosition();
+ for (int i = 0; i < groupedAggregators.size(); i++) {
+ GroupedAggregator groupedAggregator = groupedAggregators.get(i);
+ ColumnBuilder output = pageBuilder.getColumnBuilder(groupByChannels.length + i);
+ groupedAggregator.evaluate(groupId, output);
+ }
+ }
+
+ return TsBlock.wrapBlocksWithoutCopy(
+ pageBuilder.getPositionCount(),
+ new RunLengthEncodedColumn(
+ TableScanOperator.TIME_COLUMN_TEMPLATE, pageBuilder.getPositionCount()),
+ Arrays.stream(pageBuilder.getValueColumnBuilders())
+ .map(ColumnBuilder::build)
+ .toArray(Column[]::new));
+ }
+
+ public List<TSDataType> buildTypes() {
+
+ return Stream.concat(
+ groupByOutputTypes.stream().map(InternalTypeManager::getTSDataType),
+ groupedAggregators.stream().map(GroupedAggregator::getType))
+ .collect(Collectors.toList());
+ }
+
+ private Iterator<Integer> consecutiveGroupIds() {
+ return IntStream.range(0, groupByHash.getGroupCount()).iterator();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/CombineHashFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/CombineHashFunction.java
new file mode 100644
index 00000000000..2bfb427c4ea
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/CombineHashFunction.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed 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.grouped.hash;
+
+public final class CombineHashFunction {
+ private CombineHashFunction() {}
+
+ public static long getHash(long previousHashValue, long value) {
+ return (31 * previousHashValue + value);
+ }
+
+ public static void combineAllHashesWithConstant(
+ long[] hashes, int fromIndex, int toIndex, long value) {
+ // checkFromToIndex(fromIndex, toIndex, hashes.length);
+
+ for (int i = fromIndex; i < toIndex; i++) {
+ hashes[i] = (31 * hashes[i]) + value;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatGroupByHash.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatGroupByHash.java
new file mode 100644
index 00000000000..36c90d4407a
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatGroupByHash.java
@@ -0,0 +1,204 @@
+/*
+ * Licensed 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.grouped.hash;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Verify.verify;
+import static java.lang.Math.min;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.FlatHash.sumExact;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOf;
+
+// This implementation assumes arrays used in the hash are always a power of 2
+public class FlatGroupByHash implements GroupByHash {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(FlatGroupByHash.class);
+ private static final int BATCH_SIZE = 1024;
+ // Max (page value count / cumulative dictionary size) to trigger the low cardinality case
+ private static final double SMALL_DICTIONARIES_MAX_CARDINALITY_RATIO = 0.25;
+
+ private final FlatHash flatHash;
+ private final int groupByChannelCount;
+ private final boolean hasPrecomputedHash;
+
+ private long currentPageSizeInBytes;
+
+ private final Column[] currentColumns;
+ private final ColumnBuilder[] currentColumnBuilders;
+ // reusable array for computing hash batches into
+ private long[] currentHashes;
+
+ public FlatGroupByHash(
+ List<Type> hashTypes,
+ boolean hasPrecomputedHash,
+ int expectedSize,
+ UpdateMemory checkMemoryReservation) {
+ this.flatHash =
+ new FlatHash(
+ TypeUtil.getFlatHashStrategy(hashTypes),
+ hasPrecomputedHash,
+ expectedSize,
+ checkMemoryReservation);
+ this.groupByChannelCount = hashTypes.size();
+ this.hasPrecomputedHash = hasPrecomputedHash;
+
+ checkArgument(expectedSize > 0, "expectedSize must be greater than zero");
+
+ int totalChannels = hashTypes.size() + (hasPrecomputedHash ? 1 : 0);
+ this.currentColumns = new Column[totalChannels];
+ this.currentColumnBuilders = new ColumnBuilder[totalChannels];
+ }
+
+ public int getPhysicalPosition(int groupId) {
+ return flatHash.getPhysicalPosition(groupId);
+ }
+
+ @Override
+ public long getRawHash(int groupId) {
+ return flatHash.hashPosition(groupId);
+ }
+
+ @Override
+ public long getEstimatedSize() {
+ return sumExact(
+ INSTANCE_SIZE, flatHash.getEstimatedSize(), currentPageSizeInBytes, sizeOf(currentHashes));
+ }
+
+ @Override
+ public int getGroupCount() {
+ return flatHash.size();
+ }
+
+ @Override
+ public void appendValuesTo(int groupId, TsBlockBuilder pageBuilder) {
+ ColumnBuilder[] columnBuilders = currentColumnBuilders;
+ for (int i = 0; i < columnBuilders.length; i++) {
+ columnBuilders[i] = pageBuilder.getValueColumnBuilders()[i];
+ }
+ flatHash.appendTo(groupId, columnBuilders);
+ }
+
+ @Override
+ public void addPage(Column[] page) {
+ if (page[0].getPositionCount() == 0) {
+ return;
+ }
+
+ currentPageSizeInBytes = Arrays.stream(page).map(Column::getRetainedSizeInBytes).count();
+
+ Column[] columns = getColumnsFromPage(page);
+ addNonDictionaryPageWork(columns);
+ }
+
+ @Override
+ public int[] getGroupIds(Column[] page) {
+ if (page[0].getPositionCount() == 0) {
+ return new int[0];
+ }
+
+ currentPageSizeInBytes = Arrays.stream(page).map(Column::getRetainedSizeInBytes).count();
+ Column[] columns = getColumnsFromPage(page);
+
+ return getNonDictionaryPageWork(columns);
+ }
+
+ @Override
+ public int getCapacity() {
+ return flatHash.getCapacity();
+ }
+
+ private int putIfAbsent(Column[] columns, int position) {
+ return flatHash.putIfAbsent(columns, position);
+ }
+
+ private long[] getHashesBufferArray() {
+ if (currentHashes == null) {
+ currentHashes = new long[BATCH_SIZE];
+ }
+ return currentHashes;
+ }
+
+ private Column[] getColumnsFromPage(Column[] page) {
+ Column[] blocks = currentColumns;
+ checkArgument(page.length == blocks.length);
+ System.arraycopy(page, 0, blocks, 0, blocks.length);
+ return blocks;
+ }
+
+ public void addNonDictionaryPageWork(Column[] columns) {
+ int lastPosition = 0;
+
+ int positionCount = columns[0].getPositionCount();
+ checkState(lastPosition <= positionCount, "position count out of bound");
+
+ int remainingPositions = positionCount - lastPosition;
+
+ long[] hashes = getHashesBufferArray();
+ while (remainingPositions != 0) {
+ int batchSize = min(remainingPositions, hashes.length);
+ if (!flatHash.ensureAvailableCapacity(batchSize)) {
+ throw new RuntimeException("Memory for flatHash is not enough");
+ }
+
+ flatHash.computeHashes(columns, hashes, lastPosition, batchSize);
+ for (int i = 0; i < batchSize; i++) {
+ flatHash.putIfAbsent(columns, lastPosition + i, hashes[i]);
+ }
+
+ lastPosition += batchSize;
+ remainingPositions -= batchSize;
+ }
+ verify(lastPosition == positionCount);
+ }
+
+ public int[] getNonDictionaryPageWork(Column[] columns) {
+ int lastPosition = 0;
+
+ int positionCount = columns[0].getPositionCount();
+ int[] groupIds = new int[positionCount];
+ checkState(lastPosition <= positionCount, "position count out of bound");
+
+ int remainingPositions = positionCount - lastPosition;
+
+ long[] hashes = getHashesBufferArray();
+ while (remainingPositions != 0) {
+ int batchSize = min(remainingPositions, hashes.length);
+ if (!flatHash.ensureAvailableCapacity(batchSize)) {
+ throw new RuntimeException("Memory for flatHash is not enough");
+ }
+
+ flatHash.computeHashes(columns, hashes, lastPosition, batchSize);
+ for (int i = 0, position = lastPosition; i < batchSize; i++, position++) {
+ groupIds[position] = flatHash.putIfAbsent(columns, lastPosition + i, hashes[i]);
+ }
+
+ lastPosition += batchSize;
+ remainingPositions -= batchSize;
+ }
+ verify(lastPosition == positionCount);
+ return groupIds;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHash.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHash.java
new file mode 100644
index 00000000000..cbf3509c5bd
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHash.java
@@ -0,0 +1,599 @@
+/*
+ * Licensed 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.grouped.hash;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Throwables.throwIfUnchecked;
+import static com.google.common.base.Verify.verify;
+import static java.lang.Math.addExact;
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+import static java.lang.Math.toIntExact;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.VariableWidthData.EMPTY_CHUNK;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.VariableWidthData.POINTER_SIZE;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfByteArray;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfIntArray;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfObjectArray;
+
+public final class FlatHash {
+ private static final long INSTANCE_SIZE = RamUsageEstimator.shallowSizeOfInstance(FlatHash.class);
+
+ private static final double DEFAULT_LOAD_FACTOR = 15.0 / 16;
+
+ private static int computeCapacity(int maxSize, double loadFactor) {
+ int capacity = (int) (maxSize / loadFactor);
+ return max(toIntExact(1L << (64 - Long.numberOfLeadingZeros(capacity - 1))), 16);
+ }
+
+ private static final int RECORDS_PER_GROUP_SHIFT = 10;
+ private static final int RECORDS_PER_GROUP = 1 << RECORDS_PER_GROUP_SHIFT;
+ private static final int RECORDS_PER_GROUP_MASK = RECORDS_PER_GROUP - 1;
+
+ private static final int VECTOR_LENGTH = Long.BYTES;
+
+ private final FlatHashStrategy flatHashStrategy;
+ private final boolean hasPrecomputedHash;
+
+ private final int recordSize;
+ private final int recordGroupIdOffset;
+ private final int recordHashOffset;
+ private final int recordValueOffset;
+
+ private int capacity;
+ private int mask;
+
+ private byte[] control;
+ private byte[][] recordGroups;
+ private final VariableWidthData variableWidthData;
+
+ // position of each group in the hash table
+ private int[] groupRecordIndex;
+
+ // reserve enough memory before rehash
+ private final UpdateMemory checkMemoryReservation;
+ private long fixedSizeEstimate;
+ private long rehashMemoryReservation;
+
+ private int nextGroupId;
+ private int maxFill;
+
+ public FlatHash(
+ FlatHashStrategy flatHashStrategy,
+ boolean hasPrecomputedHash,
+ int expectedSize,
+ UpdateMemory checkMemoryReservation) {
+ this.flatHashStrategy = flatHashStrategy;
+ this.hasPrecomputedHash = hasPrecomputedHash;
+ this.checkMemoryReservation = checkMemoryReservation;
+
+ capacity = max(VECTOR_LENGTH, computeCapacity(expectedSize, DEFAULT_LOAD_FACTOR));
+ maxFill = calculateMaxFill(capacity);
+ mask = capacity - 1;
+ control = new byte[capacity + VECTOR_LENGTH];
+
+ groupRecordIndex = new int[maxFill];
+
+ // the record is laid out as follows:
+ // 1. optional variable width pointer
+ // 2. groupId (int)
+ // 3. fixed data for each type
+ boolean variableWidth = flatHashStrategy.isAnyVariableWidth();
+ variableWidthData = variableWidth ? new VariableWidthData() : null;
+ recordGroupIdOffset = (variableWidth ? POINTER_SIZE : 0);
+ recordHashOffset = recordGroupIdOffset + Integer.BYTES;
+ recordValueOffset = recordHashOffset + (hasPrecomputedHash ? Long.BYTES : 0);
+ recordSize = recordValueOffset + flatHashStrategy.getTotalFlatFixedLength();
+
+ recordGroups = createRecordGroups(capacity, recordSize);
+ fixedSizeEstimate = computeFixedSizeEstimate(capacity, recordSize);
+ }
+
+ public long getEstimatedSize() {
+ return sumExact(
+ fixedSizeEstimate,
+ (variableWidthData == null ? 0 : variableWidthData.getRetainedSizeBytes()),
+ rehashMemoryReservation);
+ }
+
+ public int size() {
+ return nextGroupId;
+ }
+
+ public int getCapacity() {
+ return capacity;
+ }
+
+ public static long bytesToLong(byte[] bytes, int index) {
+ if (bytes.length - 8 < 4) {
+ throw new IllegalArgumentException("Invalid input: bytes.length - offset < 8");
+ }
+ long num = 0;
+ for (int ix = index + 7; ix >= index; ix--) {
+ num <<= 8;
+ num |= (bytes[ix] & 0xff);
+ }
+ return num;
+ }
+
+ public static int bytesToInt(byte[] bytes, int index) {
+ if (bytes.length - index < 4) {
+ throw new IllegalArgumentException("Invalid input: bytes.length - offset < 4");
+ }
+
+ int num = 0;
+ for (int ix = index + 3; ix >= index; ix--) {
+ num <<= 8;
+ num |= (bytes[ix] & 0xff);
+ }
+ return num;
+ }
+
+ public static void intToBytes(byte[] desc, int offset, int i) {
+ if (desc.length - offset < 4) {
+ throw new IllegalArgumentException("Invalid input: desc.length - offset < 4");
+ }
+ desc[3 + offset] = (byte) ((i >> 24) & 0xFF);
+ desc[2 + offset] = (byte) ((i >> 16) & 0xFF);
+ desc[1 + offset] = (byte) ((i >> 8) & 0xFF);
+ desc[offset] = (byte) (i & 0xFF);
+ }
+
+ private static void longToBytes(byte[] desc, int offset, long num) {
+ if (desc.length - offset < 8) {
+ throw new IllegalArgumentException("Invalid input: desc.length - offset < 4");
+ }
+ for (int ix = 0; ix < 8; ++ix) {
+ int i = ix * 8;
+ desc[ix + offset] = (byte) ((num >> i) & 0xff);
+ }
+ }
+
+ public long hashPosition(int groupId) {
+ // for spilling
+ checkArgument(groupId < nextGroupId, "groupId out of range");
+
+ int index = groupRecordIndex[groupId];
+ byte[] records = getRecords(index);
+ if (hasPrecomputedHash) {
+ return bytesToLong(records, getRecordOffset(index) + recordHashOffset);
+ } else {
+ return valueHashCode(records, index);
+ }
+ }
+
+ public void appendTo(int groupId, ColumnBuilder[] columnBuilders) {
+ checkArgument(groupId < nextGroupId, "groupId out of range");
+ int index = groupRecordIndex[groupId];
+ byte[] records = getRecords(index);
+ int recordOffset = getRecordOffset(index);
+
+ byte[] variableWidthChunk = EMPTY_CHUNK;
+ if (variableWidthData != null) {
+ variableWidthChunk = variableWidthData.getChunk(records, recordOffset);
+ }
+
+ flatHashStrategy.readFlat(
+ records, recordOffset + recordValueOffset, variableWidthChunk, columnBuilders);
+ if (hasPrecomputedHash) {
+ columnBuilders[columnBuilders.length - 1].writeLong(
+ bytesToLong(records, recordOffset + recordHashOffset));
+ }
+ }
+
+ public boolean contains(Column[] columns, int position) {
+ return contains(columns, position, flatHashStrategy.hash(columns, position));
+ }
+
+ public boolean contains(Column[] columns, int position, long hash) {
+ return getIndex(columns, position, hash) >= 0;
+ }
+
+ public void computeHashes(Column[] columns, long[] hashes, int offset, int length) {
+ if (hasPrecomputedHash) {
+ Column hashColumn = columns[columns.length - 1];
+ for (int i = 0; i < length; i++) {
+ hashes[i] = hashColumn.getLong(offset + i);
+ }
+ } else {
+ flatHashStrategy.hashBatched(columns, hashes, offset, length);
+ }
+ }
+
+ public int putIfAbsent(Column[] columns, int position, long hash) {
+ int index = getIndex(columns, position, hash);
+ if (index >= 0) {
+ return bytesToInt(getRecords(index), getRecordOffset(index) + recordGroupIdOffset);
+ }
+
+ index = -index - 1;
+ int groupId = addNewGroup(index, columns, position, hash);
+ if (nextGroupId >= maxFill) {
+ rehash(0);
+ }
+ return groupId;
+ }
+
+ public int putIfAbsent(Column[] columns, int position) {
+ long hash;
+ /*if (hasPrecomputedHash) {
+ hash = BIGINT.getLong(columns[columns.length - 1], position);
+ } */
+
+ hash = flatHashStrategy.hash(columns, position);
+
+ return putIfAbsent(columns, position, hash);
+ }
+
+ private int getIndex(Column[] columns, int position, long hash) {
+ byte hashPrefix = (byte) (hash & 0x7F | 0x80);
+ int bucket = bucket((int) (hash >> 7));
+
+ int step = 1;
+ long repeated = repeat(hashPrefix);
+
+ while (true) {
+ final long controlVector = bytesToLong(control, bucket);
+
+ int matchIndex = matchInVector(columns, position, hash, bucket, repeated, controlVector);
+ if (matchIndex >= 0) {
+ return matchIndex;
+ }
+
+ int emptyIndex = findEmptyInVector(controlVector, bucket);
+ if (emptyIndex >= 0) {
+ return -emptyIndex - 1;
+ }
+
+ bucket = bucket(bucket + step);
+ step += VECTOR_LENGTH;
+ }
+ }
+
+ private int matchInVector(
+ Column[] columns,
+ int position,
+ long hash,
+ int vectorStartBucket,
+ long repeated,
+ long controlVector) {
+ long controlMatches = match(controlVector, repeated);
+ while (controlMatches != 0) {
+ int index = bucket(vectorStartBucket + (Long.numberOfTrailingZeros(controlMatches) >>> 3));
+ if (valueNotDistinctFrom(index, columns, position, hash)) {
+ return index;
+ }
+
+ controlMatches = controlMatches & (controlMatches - 1);
+ }
+ return -1;
+ }
+
+ private int findEmptyInVector(long vector, int vectorStartBucket) {
+ long controlMatches = match(vector, 0x00_00_00_00_00_00_00_00L);
+ if (controlMatches == 0) {
+ return -1;
+ }
+ int slot = Long.numberOfTrailingZeros(controlMatches) >>> 3;
+ return bucket(vectorStartBucket + slot);
+ }
+
+ private int addNewGroup(int index, Column[] columns, int position, long hash) {
+ setControl(index, (byte) (hash & 0x7F | 0x80));
+
+ byte[] records = getRecords(index);
+ int recordOffset = getRecordOffset(index);
+
+ int groupId = nextGroupId++;
+ intToBytes(records, recordOffset + recordGroupIdOffset, groupId);
+ groupRecordIndex[groupId] = index;
+
+ if (hasPrecomputedHash) {
+ longToBytes(records, recordOffset + recordHashOffset, hash);
+ }
+
+ byte[] variableWidthChunk = EMPTY_CHUNK;
+ int variableWidthChunkOffset = 0;
+ if (variableWidthData != null) {
+ int variableWidthSize = flatHashStrategy.getTotalVariableWidth(columns, position);
+ variableWidthChunk = variableWidthData.allocate(records, recordOffset, variableWidthSize);
+ variableWidthChunkOffset = VariableWidthData.getChunkOffset(records, recordOffset);
+ }
+ flatHashStrategy.writeFlat(
+ columns,
+ position,
+ records,
+ recordOffset + recordValueOffset,
+ variableWidthChunk,
+ variableWidthChunkOffset);
+ return groupId;
+ }
+
+ private void setControl(int index, byte hashPrefix) {
+ control[index] = hashPrefix;
+ if (index < VECTOR_LENGTH) {
+ control[index + capacity] = hashPrefix;
+ }
+ }
+
+ public boolean ensureAvailableCapacity(int batchSize) {
+ long requiredMaxFill = nextGroupId + batchSize;
+ if (requiredMaxFill >= maxFill) {
+ long minimumRequiredCapacity = (requiredMaxFill + 1) * 16 / 15;
+ return tryRehash(toIntExact(minimumRequiredCapacity));
+ }
+ return true;
+ }
+
+ private boolean tryRehash(int minimumRequiredCapacity) {
+ int newCapacity = computeNewCapacity(minimumRequiredCapacity);
+
+ // update the fixed size estimate to the new size as we will need this much memory after the
+ // rehash
+ fixedSizeEstimate = computeFixedSizeEstimate(newCapacity, recordSize);
+
+ // the rehash incrementally allocates the new records as needed, so as new memory is added old
+ // memory is released
+ // while the rehash is in progress, the old control array is retained, and one additional record
+ // group is retained
+ rehashMemoryReservation = sumExact(sizeOf(control), sizeOf(recordGroups[0]));
+ verify(rehashMemoryReservation >= 0, "rehashMemoryReservation is negative");
+ if (!checkMemoryReservation.update()) {
+ return false;
+ }
+
+ rehash(minimumRequiredCapacity);
+ return true;
+ }
+
+ private void rehash(int minimumRequiredCapacity) {
+ int oldCapacity = capacity;
+ byte[] oldControl = control;
+ byte[][] oldRecordGroups = recordGroups;
+
+ capacity = computeNewCapacity(minimumRequiredCapacity);
+ maxFill = calculateMaxFill(capacity);
+ mask = capacity - 1;
+
+ control = new byte[capacity + VECTOR_LENGTH];
+
+ // we incrementally allocate the record groups to smooth out memory allocation
+ if (capacity <= RECORDS_PER_GROUP) {
+ recordGroups = new byte[][] {new byte[multiplyExact(capacity, recordSize)]};
+ } else {
+ recordGroups = new byte[(capacity + 1) >> RECORDS_PER_GROUP_SHIFT][];
+ }
+
+ groupRecordIndex = new int[maxFill];
+
+ for (int oldRecordGroupIndex = 0;
+ oldRecordGroupIndex < oldRecordGroups.length;
+ oldRecordGroupIndex++) {
+ byte[] oldRecords = oldRecordGroups[oldRecordGroupIndex];
+ oldRecordGroups[oldRecordGroupIndex] = null;
+ for (int indexInRecordGroup = 0;
+ indexInRecordGroup < min(RECORDS_PER_GROUP, oldCapacity);
+ indexInRecordGroup++) {
+ int oldIndex = (oldRecordGroupIndex << RECORDS_PER_GROUP_SHIFT) + indexInRecordGroup;
+ if (oldControl[oldIndex] == 0) {
+ continue;
+ }
+
+ long hash;
+ if (hasPrecomputedHash) {
+ hash = bytesToLong(oldRecords, getRecordOffset(oldIndex) + recordHashOffset);
+ } else {
+ hash = valueHashCode(oldRecords, oldIndex);
+ }
+
+ byte hashPrefix = (byte) (hash & 0x7F | 0x80);
+ int bucket = bucket((int) (hash >> 7));
+
+ // getIndex is not used here because values in a rehash are always distinct
+ int step = 1;
+ while (true) {
+ final long controlVector = bytesToLong(control, bucket);
+ // values are already distinct, so just find the first empty slot
+ int emptyIndex = findEmptyInVector(controlVector, bucket);
+ if (emptyIndex >= 0) {
+ setControl(emptyIndex, hashPrefix);
+
+ int newRecordGroupIndex = emptyIndex >> RECORDS_PER_GROUP_SHIFT;
+ byte[] records = recordGroups[newRecordGroupIndex];
+ if (records == null) {
+ records = new byte[multiplyExact(RECORDS_PER_GROUP, recordSize)];
+ recordGroups[newRecordGroupIndex] = records;
+ }
+ int recordOffset = getRecordOffset(emptyIndex);
+ int oldRecordOffset = getRecordOffset(oldIndex);
+ System.arraycopy(oldRecords, oldRecordOffset, records, recordOffset, recordSize);
+
+ int groupId = bytesToInt(records, recordOffset + recordGroupIdOffset);
+ groupRecordIndex[groupId] = emptyIndex;
+
+ break;
+ }
+
+ bucket = bucket(bucket + step);
+ step += VECTOR_LENGTH;
+ }
+ }
+ }
+
+ // add any completely empty record groups
+ // the odds of needing this are exceedingly low, but it is technically possible
+ for (int i = 0; i < recordGroups.length; i++) {
+ if (recordGroups[i] == null) {
+ recordGroups[i] = new byte[multiplyExact(RECORDS_PER_GROUP, recordSize)];
+ }
+ }
+
+ // release temporary memory reservation
+ rehashMemoryReservation = 0;
+ fixedSizeEstimate = computeFixedSizeEstimate(capacity, recordSize);
+ checkMemoryReservation.update();
+ }
+
+ private int computeNewCapacity(int minimumRequiredCapacity) {
+ checkArgument(minimumRequiredCapacity >= 0, "minimumRequiredCapacity must be positive");
+ long newCapacityLong = capacity * 2L;
+ while (newCapacityLong < minimumRequiredCapacity) {
+ newCapacityLong = multiplyExact(newCapacityLong, 2);
+ }
+ if (newCapacityLong > Integer.MAX_VALUE) {
+ throw new RuntimeException("Size of hash table cannot exceed 1 billion entries");
+ }
+ return toIntExact(newCapacityLong);
+ }
+
+ // copy from JDK
+ public static long multiplyExact(long x, long y) {
+ long r = x * y;
+ long ax = Math.abs(x);
+ long ay = Math.abs(y);
+ if (((ax | ay) >>> 31 != 0)) {
+ // Some bits greater than 2^31 that might cause overflow
+ // Check the result using the divide operator
+ // and check for the special case of Long.MIN_VALUE * -1
+ if (((y != 0) && (r / y != x)) || (x == Long.MIN_VALUE && y == -1)) {
+ throw new ArithmeticException("long overflow");
+ }
+ }
+ return r;
+ }
+
+ // copy from JDK
+ public static int multiplyExact(int x, int y) {
+ long r = (long) x * (long) y;
+ if ((int) r != r) {
+ throw new ArithmeticException("integer overflow");
+ }
+ return (int) r;
+ }
+
+ private int bucket(int hash) {
+ return hash & mask;
+ }
+
+ private byte[] getRecords(int index) {
+ return recordGroups[index >> RECORDS_PER_GROUP_SHIFT];
+ }
+
+ private int getRecordOffset(int index) {
+ return (index & RECORDS_PER_GROUP_MASK) * recordSize;
+ }
+
+ private long valueHashCode(byte[] records, int index) {
+ int recordOffset = getRecordOffset(index);
+
+ try {
+ byte[] variableWidthChunk = EMPTY_CHUNK;
+ if (variableWidthData != null) {
+ variableWidthChunk = variableWidthData.getChunk(records, recordOffset);
+ }
+
+ return flatHashStrategy.hash(records, recordOffset + recordValueOffset, variableWidthChunk);
+ } catch (Throwable throwable) {
+ throwIfUnchecked(throwable);
+ throw new RuntimeException(throwable);
+ }
+ }
+
+ private boolean valueNotDistinctFrom(
+ int leftIndex, Column[] rightColumns, int rightPosition, long rightHash) {
+ byte[] leftRecords = getRecords(leftIndex);
+ int leftRecordOffset = getRecordOffset(leftIndex);
+
+ if (hasPrecomputedHash) {
+ long leftHash = bytesToLong(leftRecords, leftRecordOffset + recordHashOffset);
+ if (leftHash != rightHash) {
+ return false;
+ }
+ }
+
+ byte[] leftVariableWidthChunk = EMPTY_CHUNK;
+ if (variableWidthData != null) {
+ leftVariableWidthChunk = variableWidthData.getChunk(leftRecords, leftRecordOffset);
+ }
+
+ return flatHashStrategy.valueNotDistinctFrom(
+ leftRecords,
+ leftRecordOffset + recordValueOffset,
+ leftVariableWidthChunk,
+ rightColumns,
+ rightPosition);
+ }
+
+ private static long repeat(byte value) {
+ return ((value & 0xFF) * 0x01_01_01_01_01_01_01_01L);
+ }
+
+ private static long match(long vector, long repeatedValue) {
+ // HD 6-1
+ long comparison = vector ^ repeatedValue;
+ return (comparison - 0x01_01_01_01_01_01_01_01L) & ~comparison & 0x80_80_80_80_80_80_80_80L;
+ }
+
+ public int getPhysicalPosition(int groupId) {
+ return groupRecordIndex[groupId];
+ }
+
+ private static int calculateMaxFill(int capacity) {
+ return toIntExact(capacity * 15L / 16);
+ }
+
+ private static byte[][] createRecordGroups(int capacity, int recordSize) {
+ if (capacity <= RECORDS_PER_GROUP) {
+ return new byte[][] {new byte[multiplyExact(capacity, recordSize)]};
+ }
+
+ byte[][] groups = new byte[(capacity + 1) >> RECORDS_PER_GROUP_SHIFT][];
+ for (int i = 0; i < groups.length; i++) {
+ groups[i] = new byte[multiplyExact(RECORDS_PER_GROUP, recordSize)];
+ }
+ return groups;
+ }
+
+ private static long computeRecordGroupsSize(int capacity, int recordSize) {
+ if (capacity <= RECORDS_PER_GROUP) {
+ return sizeOfObjectArray(1) + sizeOfByteArray(multiplyExact(capacity, recordSize));
+ }
+
+ int groupCount = addExact(capacity, 1) >> RECORDS_PER_GROUP_SHIFT;
+ return sizeOfObjectArray(groupCount)
+ + multiplyExact(groupCount, sizeOfByteArray(multiplyExact(RECORDS_PER_GROUP, recordSize)));
+ }
+
+ private static long computeFixedSizeEstimate(int capacity, int recordSize) {
+ return sumExact(
+ INSTANCE_SIZE,
+ sizeOfByteArray(capacity + VECTOR_LENGTH),
+ computeRecordGroupsSize(capacity, recordSize),
+ sizeOfIntArray(capacity));
+ }
+
+ public static long sumExact(long... values) {
+ long result = 0;
+ for (long value : values) {
+ result = addExact(result, value);
+ }
+ return result;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHashStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHashStrategy.java
new file mode 100644
index 00000000000..3dbd09b3542
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/FlatHashStrategy.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed 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.grouped.hash;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+public interface FlatHashStrategy {
+ boolean isAnyVariableWidth();
+
+ int getTotalFlatFixedLength();
+
+ int getTotalVariableWidth(Column[] columns, int position);
+
+ void readFlat(
+ byte[] fixedChunk, int fixedOffset, byte[] variableChunk, ColumnBuilder[] columnBuilders);
+
+ void writeFlat(
+ Column[] columns,
+ int position,
+ byte[] fixedChunk,
+ int fixedOffset,
+ byte[] variableChunk,
+ int variableOffset);
+
+ boolean valueNotDistinctFrom(
+ byte[] leftFixedChunk,
+ int leftFixedOffset,
+ byte[] leftVariableChunk,
+ Column[] rightColumns,
+ int rightPosition);
+
+ long hash(Column[] columns, int position);
+
+ long hash(byte[] fixedChunk, int fixedOffset, byte[] variableChunk);
+
+ void hashBatched(Column[] columns, long[] hashes, int offset, int length);
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/GroupByHash.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/GroupByHash.java
new file mode 100644
index 00000000000..2ee9f15e28d
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/GroupByHash.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed 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.grouped.hash;
+
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.type.Type;
+
+import java.util.List;
+
+public interface GroupByHash {
+ static GroupByHash createGroupByHash(
+ List<Type> types, boolean hasPrecomputedHash, int expectedSize, UpdateMemory updateMemory) {
+ /*if (types.size() == 1 && types.get(0).equals(BIGINT)) {
+ return new BigintGroupByHash(hasPrecomputedHash, expectedSize, updateMemory);
+ }*/
+ return new FlatGroupByHash(types, hasPrecomputedHash, expectedSize, updateMemory);
+ }
+
+ long getEstimatedSize();
+
+ int getGroupCount();
+
+ void appendValuesTo(int groupId, TsBlockBuilder pageBuilder);
+
+ void addPage(Column[] groupedColumns);
+
+ /**
+ * The order of new group ids need to be the same as the order of incoming rows, i.e. new group
+ * ids should be assigned in rows iteration order Example: rows: A B C B D A E group ids: 1 2 3 2
+ * 4 1 5
+ */
+ int[] getGroupIds(Column[] groupedColumns);
+
+ long getRawHash(int groupId);
+
+ @VisibleForTesting
+ int getCapacity();
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/HashStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/HashStrategy.java
new file mode 100644
index 00000000000..bd6a4a52ebe
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/HashStrategy.java
@@ -0,0 +1,225 @@
+/*
+ * 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.grouped.hash;
+
+import org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.type.Type;
+
+import java.util.List;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.getFlatVariableWidthSize;
+import static org.apache.iotdb.db.queryengine.plan.relational.utils.TypeUtil.notDistinctFrom;
+
+public class HashStrategy implements FlatHashStrategy {
+
+ private final boolean isAnyVariableWidth;
+
+ private final int totalFlatFixedLength;
+
+ private final List<KeyField> keyFields;
+
+ public HashStrategy(List<Type> types) {
+ this.isAnyVariableWidth = types.stream().anyMatch(TypeUtil::isFlatVariableWidth);
+
+ int fixedOffset = 0;
+ ImmutableList.Builder<KeyField> builder = ImmutableList.builder();
+ for (int i = 0; i < types.size(); i++) {
+ Type type = types.get(i);
+ builder.add(new KeyField(i, type, fixedOffset));
+ fixedOffset += 1 + TypeUtil.getFlatFixedSize(type);
+ }
+ keyFields = builder.build();
+
+ totalFlatFixedLength = fixedOffset;
+ }
+
+ @Override
+ public boolean isAnyVariableWidth() {
+ return isAnyVariableWidth;
+ }
+
+ @Override
+ public int getTotalFlatFixedLength() {
+ return totalFlatFixedLength;
+ }
+
+ @Override
+ public int getTotalVariableWidth(Column[] columns, int position) {
+ int result = 0;
+ for (int i = 0; i < columns.length; i++) {
+ result += getFlatVariableWidthSize(keyFields.get(i).type, columns[i], position);
+ }
+ return result;
+ }
+
+ @Override
+ public void readFlat(
+ byte[] fixedChunk, int fixedOffset, byte[] variableChunk, ColumnBuilder[] columnBuilders) {
+ for (int i = 0; i < columnBuilders.length; i++) {
+ KeyField keyField = keyFields.get(i);
+ if (fixedChunk[fixedOffset + keyField.fieldIsNullOffset] != 0) {
+ columnBuilders[i].appendNull();
+ } else {
+ TypeUtil.readFlat(
+ keyField.type,
+ fixedChunk,
+ fixedOffset + keyField.fieldFixedOffset,
+ variableChunk,
+ columnBuilders[i]);
+ }
+ }
+ }
+
+ @Override
+ public void writeFlat(
+ Column[] columns,
+ int position,
+ byte[] fixedChunk,
+ int fixedOffset,
+ byte[] variableChunk,
+ int variableOffset) {
+ for (int i = 0; i < columns.length; i++) {
+ KeyField keyField = keyFields.get(i);
+ if (columns[i].isNull(position)) {
+ fixedChunk[fixedOffset + keyField.fieldIsNullOffset] = (byte) 1;
+ } else {
+ TypeUtil.writeFlat(
+ keyField.getType(),
+ columns[i],
+ position,
+ fixedChunk,
+ fixedOffset + keyField.fieldFixedOffset,
+ variableChunk,
+ variableOffset);
+ variableOffset += TypeUtil.getFlatVariableWidthSize(keyField.type, columns[i], position);
+ }
+ }
+ }
+
+ @Override
+ public boolean valueNotDistinctFrom(
+ byte[] leftFixedChunk,
+ int leftFixedOffset,
+ byte[] leftVariableChunk,
+ Column[] rightColumns,
+ int rightPosition) {
+ for (int i = 0; i < rightColumns.length; i++) {
+ KeyField keyField = keyFields.get(i);
+ boolean leftIsNull = leftFixedChunk[leftFixedOffset + keyField.fieldIsNullOffset] != 0;
+ boolean rightIsNull = rightColumns[i].isNull(rightPosition);
+
+ if (leftIsNull != rightIsNull) {
+ return false;
+ }
+
+ if (leftIsNull) {
+ continue;
+ }
+
+ if (!notDistinctFrom(
+ keyField.type,
+ leftFixedChunk,
+ leftFixedOffset + keyField.fieldFixedOffset,
+ leftVariableChunk,
+ rightColumns[i],
+ rightPosition)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public long hash(Column[] columns, int position) {
+ long result = 0L;
+ long hash;
+ for (int i = 0; i < columns.length; i++) {
+ if (columns[i].isNull(position)) {
+ hash = 0L;
+ } else {
+ hash = TypeUtil.hash(keyFields.get(i).type, columns[i], position);
+ }
+ result = CombineHashFunction.getHash(result, hash);
+ }
+
+ return result;
+ }
+
+ @Override
+ public long hash(byte[] fixedChunk, int fixedOffset, byte[] variableChunk) {
+ long result = 0L;
+ long hash;
+ for (int i = 0; i < keyFields.size(); i++) {
+ if (fixedChunk[fixedOffset + keyFields.get(i).fieldIsNullOffset] != 0) {
+ hash = 0L;
+ } else {
+ hash =
+ TypeUtil.hash(
+ keyFields.get(i).getType(),
+ fixedChunk,
+ fixedOffset + keyFields.get(i).fieldFixedOffset,
+ variableChunk);
+ }
+ result = CombineHashFunction.getHash(result, hash);
+ }
+
+ return result;
+ }
+
+ @Override
+ public void hashBatched(Column[] columns, long[] hashes, int offset, int length) {
+ for (int i = 0; i < length; i++) {
+ hashes[i] = hash(columns, i);
+ }
+ }
+
+ private static class KeyField {
+ private final int index;
+ private final Type type;
+ private final int fieldIsNullOffset;
+ private final int fieldFixedOffset;
+
+ KeyField(int index, Type type, int fieldIsNullOffset) {
+ this.index = index;
+ this.type = type;
+ this.fieldIsNullOffset = fieldIsNullOffset;
+ fieldFixedOffset = fieldIsNullOffset + 1;
+ }
+
+ public int getIndex() {
+ return index;
+ }
+
+ public Type getType() {
+ return type;
+ }
+
+ public int getFieldIsNullOffset() {
+ return fieldIsNullOffset;
+ }
+
+ public int getFieldFixedOffset() {
+ return fieldFixedOffset;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/VariableWidthData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/VariableWidthData.java
new file mode 100644
index 00000000000..47c27a8219f
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/VariableWidthData.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed 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.grouped.hash;
+
+import com.google.common.primitives.Ints;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static com.google.common.base.Verify.verify;
+import static java.lang.Math.addExact;
+import static java.lang.Math.max;
+import static java.lang.Math.subtractExact;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.FlatHash.bytesToInt;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.FlatHash.intToBytes;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.FlatHash.sumExact;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOf;
+import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfObjectArray;
+
+public final class VariableWidthData {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(VariableWidthData.class);
+
+ public static final int MIN_CHUNK_SIZE = 1024;
+ public static final int MAX_CHUNK_SIZE = 8 * 1024 * 1024;
+
+ public static final int POINTER_SIZE = Integer.BYTES + Integer.BYTES + Integer.BYTES;
+
+ public static final byte[] EMPTY_CHUNK = new byte[0];
+
+ private final List<byte[]> chunks = new ArrayList<>();
+ private int openChunkOffset;
+
+ private long chunksRetainedSizeInBytes;
+
+ private long allocatedBytes;
+ private long freeBytes;
+
+ public VariableWidthData() {}
+
+ public VariableWidthData(VariableWidthData variableWidthData) {
+ for (byte[] chunk : variableWidthData.chunks) {
+ chunks.add(Arrays.copyOf(chunk, chunk.length));
+ }
+ this.openChunkOffset = variableWidthData.openChunkOffset;
+
+ this.chunksRetainedSizeInBytes = variableWidthData.chunksRetainedSizeInBytes;
+
+ this.allocatedBytes = variableWidthData.allocatedBytes;
+ this.freeBytes = variableWidthData.freeBytes;
+ }
+
+ public VariableWidthData(List<byte[]> chunks, int openChunkOffset) {
+ this.chunks.addAll(chunks);
+ this.openChunkOffset = openChunkOffset;
+ this.chunksRetainedSizeInBytes =
+ chunks.stream().mapToLong(RamUsageEstimator::sizeOf).reduce(0L, Math::addExact);
+ this.allocatedBytes = chunks.stream().mapToLong(chunk -> chunk.length).sum();
+ this.freeBytes = 0;
+ }
+
+ public long getRetainedSizeBytes() {
+ return sumExact(INSTANCE_SIZE, chunksRetainedSizeInBytes, sizeOfObjectArray(chunks.size()));
+ }
+
+ public List<byte[]> getAllChunks() {
+ return chunks;
+ }
+
+ public long getAllocatedBytes() {
+ return allocatedBytes;
+ }
+
+ public long getFreeBytes() {
+ return freeBytes;
+ }
+
+ public byte[] allocate(byte[] pointer, int pointerOffset, int size) {
+ if (size == 0) {
+ writePointer(pointer, pointerOffset, 0, 0, 0);
+ return EMPTY_CHUNK;
+ }
+
+ byte[] openChunk = chunks.isEmpty() ? EMPTY_CHUNK : chunks.get(chunks.size() - 1);
+ if (openChunk.length - openChunkOffset < size) {
+ // record unused space as free bytes
+ freeBytes += (openChunk.length - openChunkOffset);
+
+ // allocate enough space for 32 values of the current size, or double the current chunk size,
+ // whichever is larger
+ int newSize = Ints.saturatedCast(max(size * 32L, openChunk.length * 2L));
+ // constrain to be between min and max chunk size
+ newSize = clamp(newSize, MIN_CHUNK_SIZE, MAX_CHUNK_SIZE);
+ // jumbo rows get a separate allocation
+ newSize = max(newSize, size);
+ openChunk = new byte[newSize];
+ chunks.add(openChunk);
+ allocatedBytes += newSize;
+ chunksRetainedSizeInBytes = addExact(chunksRetainedSizeInBytes, sizeOf(openChunk));
+ openChunkOffset = 0;
+ }
+
+ writePointer(pointer, pointerOffset, chunks.size() - 1, openChunkOffset, size);
+ openChunkOffset += size;
+ return openChunk;
+ }
+
+ public static int clamp(long value, int min, int max) {
+ if (min > max) {
+ throw new IllegalArgumentException(min + " > " + max);
+ }
+ return (int) Math.min(max, Math.max(value, min));
+ }
+
+ public void free(byte[] pointer, int pointerOffset) {
+ int valueLength = getValueLength(pointer, pointerOffset);
+ if (valueLength == 0) {
+ return;
+ }
+
+ int valueChunkIndex = getChunkIndex(pointer, pointerOffset);
+ byte[] valueChunk = chunks.get(valueChunkIndex);
+
+ // if this is the last value in the open byte[], then we can simply back up the open chunk
+ // offset
+ if (valueChunkIndex == chunks.size() - 1) {
+ int valueOffset = getChunkOffset(pointer, pointerOffset);
+ if (this.openChunkOffset - valueLength == valueOffset) {
+ this.openChunkOffset = valueOffset;
+ return;
+ }
+ }
+
+ // if this is the only value written to the chunk, we can simply replace the chunk with the
+ // empty chunk
+ if (valueLength == valueChunk.length) {
+ chunks.set(valueChunkIndex, EMPTY_CHUNK);
+ chunksRetainedSizeInBytes = subtractExact(chunksRetainedSizeInBytes, sizeOf(valueChunk));
+ allocatedBytes -= valueChunk.length;
+ return;
+ }
+
+ freeBytes += valueLength;
+ }
+
+ public byte[] getChunk(byte[] pointer, int pointerOffset) {
+ int chunkIndex = getChunkIndex(pointer, pointerOffset);
+ if (chunks.isEmpty()) {
+ verify(chunkIndex == 0);
+ return EMPTY_CHUNK;
+ }
+ checkIndex(chunkIndex, chunks.size());
+ return chunks.get(chunkIndex);
+ }
+
+ private void checkIndex(int index, int length) {
+ if (index < 0 || index >= length) throw new IndexOutOfBoundsException();
+ }
+
+ private static int getChunkIndex(byte[] pointer, int pointerOffset) {
+ return bytesToInt(pointer, pointerOffset);
+ }
+
+ public static int getChunkOffset(byte[] pointer, int pointerOffset) {
+ return bytesToInt(pointer, pointerOffset + Integer.BYTES);
+ }
+
+ public static int getValueLength(byte[] pointer, int pointerOffset) {
+ return bytesToInt(pointer, pointerOffset + Long.BYTES);
+ }
+
+ public static void writePointer(
+ byte[] pointer, int pointerOffset, int chunkIndex, int chunkOffset, int valueLength) {
+ intToBytes(pointer, pointerOffset, chunkIndex);
+ intToBytes(pointer, pointerOffset + Integer.BYTES, chunkOffset);
+ intToBytes(pointer, pointerOffset + Long.BYTES, valueLength);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/XxHash64.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/XxHash64.java
new file mode 100644
index 00000000000..bf38564ee06
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/hash/XxHash64.java
@@ -0,0 +1,276 @@
+/*
+ * 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.grouped.hash;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import static java.lang.Long.rotateLeft;
+import static org.apache.tsfile.utils.BytesUtils.bytesToInt;
+import static org.apache.tsfile.utils.BytesUtils.bytesToLongFromOffset;
+
+public class XxHash64 {
+ private static final long PRIME64_1 = 0x9E3779B185EBCA87L;
+ private static final long PRIME64_2 = 0x9E3779B185EBCA87L;
+ private static final long PRIME64_3 = 0x165667B19E3779F9L;
+ private static final long PRIME64_4 = 0x85EBCA77C2b2AE63L;
+ private static final long PRIME64_5 = 0x27D4EB2F165667C5L;
+ private static final long DEFAULT_SEED = 0L;
+ private final long seed;
+ private static final int BUFFER_ADDRESS;
+ private final byte[] buffer = new byte[32];
+ private int bufferSize;
+ private long bodyLength;
+ private long v1;
+ private long v2;
+ private long v3;
+ private long v4;
+
+ public static final long TRUE_XX_HASH = XxHash64.hash(1);
+ public static final long FALSE_XX_HASH = XxHash64.hash(0);
+
+ public XxHash64() {
+ this(0L);
+ }
+
+ // copy from
+ // https://github.com/airlift/slice/blob/master/src/main/java/io/airlift/slice/XxHash64.java
+ public XxHash64(long seed) {
+ this.seed = seed;
+ this.v1 = seed + PRIME64_1 + PRIME64_2;
+ this.v2 = seed + PRIME64_2;
+ this.v3 = seed;
+ this.v4 = seed - PRIME64_1;
+ }
+
+ public XxHash64 update(byte[] data) {
+ return this.update(data, 0, data.length);
+ }
+
+ public XxHash64 update(byte[] data, int offset, int length) {
+ // Objects.checkFromIndexSize(offset, length, data.length);
+ this.updateHash(data, BUFFER_ADDRESS + offset, length);
+ return this;
+ }
+
+ public long hash() {
+ long hash;
+ if (this.bodyLength > 0L) {
+ hash = this.computeBody();
+ } else {
+ hash = this.seed + PRIME64_5;
+ }
+
+ hash += this.bodyLength + (long) this.bufferSize;
+ return updateTail(hash, this.buffer, BUFFER_ADDRESS, 0, this.bufferSize);
+ }
+
+ private long computeBody() {
+ long hash =
+ rotateLeft(this.v1, 1)
+ + rotateLeft(this.v2, 7)
+ + rotateLeft(this.v3, 12)
+ + rotateLeft(this.v4, 18);
+ hash = update(hash, this.v1);
+ hash = update(hash, this.v2);
+ hash = update(hash, this.v3);
+ hash = update(hash, this.v4);
+ return hash;
+ }
+
+ private void updateHash(byte[] base, int address, int length) {
+ int index;
+ if (this.bufferSize > 0) {
+ index = Math.min(32 - this.bufferSize, length);
+ System.arraycopy(base, address, this.buffer, BUFFER_ADDRESS + this.bufferSize, index);
+ this.bufferSize += index;
+ address += index;
+ length -= index;
+ if (this.bufferSize == 32) {
+ this.updateBody(this.buffer, BUFFER_ADDRESS, this.bufferSize);
+ this.bufferSize = 0;
+ }
+ }
+
+ if (length >= 32) {
+ index = this.updateBody(base, address, length);
+ address += index;
+ length -= index;
+ }
+
+ if (length > 0) {
+ System.arraycopy(base, address, this.buffer, BUFFER_ADDRESS, length);
+ this.bufferSize = length;
+ }
+ }
+
+ private int updateBody(byte[] base, int address, int length) {
+ int remaining;
+ for (remaining = length; remaining >= 32; remaining -= 32) {
+ this.v1 = mix(this.v1, bytesToLongFromOffset(base, Long.BYTES, address));
+ this.v2 = mix(this.v2, bytesToLongFromOffset(base, Long.BYTES, address + 8));
+ this.v3 = mix(this.v3, bytesToLongFromOffset(base, Long.BYTES, address + 16));
+ this.v4 = mix(this.v4, bytesToLongFromOffset(base, Long.BYTES, address + 24));
+ address += 32;
+ }
+
+ int index = length - remaining;
+ this.bodyLength += index;
+ return index;
+ }
+
+ public static long hash(long value) {
+ return hash(0L, value);
+ }
+
+ public static long hash(long seed, long value) {
+ long hash = seed + PRIME64_5 + 8L;
+ hash = updateTail(hash, value);
+ hash = finalShuffle(hash);
+ return hash;
+ }
+
+ public static long hash(InputStream in) throws IOException {
+ return hash(0L, in);
+ }
+
+ public static long hash(long seed, InputStream in) throws IOException {
+ XxHash64 hash = new XxHash64(seed);
+ byte[] buffer = new byte[8192];
+
+ while (true) {
+ int length = in.read(buffer);
+ if (length == -1) {
+ return hash.hash();
+ }
+
+ hash.update(buffer, 0, length);
+ }
+ }
+
+ public static long hash(byte[] data) {
+ return hash(data, 0, data.length);
+ }
+
+ public static long hash(long seed, byte[] data) {
+ return hash(seed, data, 0, data.length);
+ }
+
+ public static long hash(byte[] data, int offset, int length) {
+ return hash(0L, data, offset, length);
+ }
+
+ public static long hash(long seed, byte[] data, int offset, int length) {
+ // Objects.checkFromIndexSize(offset, length, data.length());
+ int address = BUFFER_ADDRESS + offset;
+ long hash;
+ if (length >= 32) {
+ hash = updateBody(seed, data, address, length);
+ } else {
+ hash = seed + PRIME64_5;
+ }
+
+ hash += length;
+ // round to the closest 32 byte boundary
+ // this is the point up to which updateBody() processed
+ int index = length & 0xFFFFFFE0;
+ return updateTail(hash, data, address, index, length);
+ }
+
+ private static long updateTail(long hash, byte[] base, int address, int index, int length) {
+ while (index <= length - 8) {
+ hash = updateTail(hash, bytesToLongFromOffset(base, Long.BYTES, address + index));
+ index += 8;
+ }
+
+ if (index <= length - 4) {
+ hash = updateTail(hash, bytesToInt(base, address + index));
+ index += 4;
+ }
+
+ while (index < length) {
+ hash = updateTail(hash, base[address + index]);
+ ++index;
+ }
+
+ hash = finalShuffle(hash);
+ return hash;
+ }
+
+ private static long updateBody(long seed, byte[] base, int address, int length) {
+ long v1 = seed + PRIME64_1 + PRIME64_2;
+ long v2 = seed + PRIME64_2;
+ long v3 = seed;
+ long v4 = seed - PRIME64_1;
+
+ for (int remaining = length; remaining >= 32; remaining -= 32) {
+ v1 = mix(v1, bytesToLongFromOffset(base, Long.BYTES, address));
+ v2 = mix(v2, bytesToLongFromOffset(base, Long.BYTES, address + 8));
+ v3 = mix(v3, bytesToLongFromOffset(base, Long.BYTES, address + 16));
+ v4 = mix(v4, bytesToLongFromOffset(base, Long.BYTES, address + 24));
+ address += 32;
+ }
+
+ long hash = rotateLeft(v1, 1) + rotateLeft(v2, 7) + rotateLeft(v3, 12) + rotateLeft(v4, 18);
+ hash = update(hash, v1);
+ hash = update(hash, v2);
+ hash = update(hash, v3);
+ hash = update(hash, v4);
+ return hash;
+ }
+
+ private static long mix(long current, long value) {
+ return rotateLeft(current + value * PRIME64_2, 31) * PRIME64_1;
+ }
+
+ private static long update(long hash, long value) {
+ long temp = hash ^ mix(0, value);
+ return temp * PRIME64_1 + PRIME64_4;
+ }
+
+ private static long updateTail(long hash, long value) {
+ long temp = hash ^ mix(0, value);
+ return rotateLeft(temp, 27) * PRIME64_1 + PRIME64_4;
+ }
+
+ private static long updateTail(long hash, int value) {
+ long unsigned = value & 0xFFFF_FFFFL;
+ long temp = hash ^ (unsigned * PRIME64_1);
+ return rotateLeft(temp, 23) * PRIME64_2 + PRIME64_3;
+ }
+
+ private static long updateTail(long hash, byte value) {
+ int unsigned = value & 0xFF;
+ long temp = hash ^ (unsigned * PRIME64_5);
+ return rotateLeft(temp, 11) * PRIME64_1;
+ }
+
+ private static long finalShuffle(long hash) {
+ hash ^= hash >>> 33;
+ hash *= PRIME64_2;
+ hash ^= hash >>> 29;
+ hash *= PRIME64_3;
+ hash ^= hash >>> 32;
+ return hash;
+ }
+
+ static {
+ BUFFER_ADDRESS = 0;
+ }
+}
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 7c749154941..718b0958f5f 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
@@ -75,6 +75,9 @@ import org.apache.iotdb.db.queryengine.execution.operator.source.relational.Tabl
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+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.GroupedAggregator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.HashAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
@@ -164,13 +167,14 @@ import java.util.Set;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.collect.ImmutableList.toImmutableList;
import static java.util.Objects.requireNonNull;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.MEASUREMENT;
-import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TIME;
import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNode;
import static org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator.getComparatorForTable;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.constructAlignedPath;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AccumulatorFactory.createAccumulator;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AccumulatorFactory.createGroupedAccumulator;
import static org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
import static org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator.ASC_TIME_COMPARATOR;
import static org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator.IDENTITY_FILL;
@@ -1319,8 +1323,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return planGlobalAggregation(node, child, context.getTypeProvider(), operatorContext);
}
- throw new UnsupportedOperationException();
- // return planGroupByAggregation(node, child, outputTypes, operatorContext);
+ return planGroupByAggregation(node, child, context.getTypeProvider(), operatorContext);
}
private Operator planGlobalAggregation(
@@ -1383,6 +1386,92 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
OptionalInt.empty());
}
+ private Operator planGroupByAggregation(
+ AggregationNode node,
+ Operator child,
+ TypeProvider typeProvider,
+ OperatorContext operatorContext) {
+ ImmutableList.Builder<GroupedAggregator> aggregatorBuilder = new ImmutableList.Builder<>();
+ Map<Symbol, Integer> childLayout =
+ makeLayoutFromOutputSymbols(node.getChild().getOutputSymbols());
+
+ node.getAggregations()
+ .forEach(
+ (k, v) ->
+ aggregatorBuilder.add(
+ buildGroupByAggregator(childLayout, k, v, node.getStep(), typeProvider)));
+
+ List<Integer> groupByChannels = getChannelsForSymbols(node.getGroupingKeys(), childLayout);
+ List<Type> groupByTypes =
+ node.getGroupingKeys().stream()
+ .map(typeProvider::getTableModelType)
+ .collect(toImmutableList());
+
+ if (node.isStreamable()) {
+ /*return new StreamingAggregationOperator;*/
+ }
+
+ return new HashAggregationOperator(
+ operatorContext,
+ child,
+ groupByTypes,
+ groupByChannels,
+ aggregatorBuilder.build(),
+ node.getStep(),
+ 10_000,
+ Long.MAX_VALUE,
+ false,
+ Long.MAX_VALUE);
+ }
+
+ private static List<Integer> getChannelsForSymbols(
+ List<Symbol> symbols, Map<Symbol, Integer> layout) {
+ ImmutableList.Builder<Integer> builder = ImmutableList.builder();
+ for (Symbol symbol : symbols) {
+ builder.add(layout.get(symbol));
+ }
+ return builder.build();
+ }
+
+ private GroupedAggregator buildGroupByAggregator(
+ Map<Symbol, Integer> childLayout,
+ Symbol symbol,
+ AggregationNode.Aggregation aggregation,
+ AggregationNode.Step step,
+ TypeProvider typeProvider) {
+ List<Integer> argumentChannels = new ArrayList<>();
+ List<TSDataType> argumentTypes = new ArrayList<>();
+ for (Expression argument : aggregation.getArguments()) {
+ Symbol argumentSymbol = Symbol.from(argument);
+ argumentChannels.add(childLayout.get(argumentSymbol));
+
+ // get argument types
+ Type type = typeProvider.getTableModelType(argumentSymbol);
+ if (type instanceof RowType) {
+ type.getTypeParameters().forEach(subType -> argumentTypes.add(getTSDataType(subType)));
+ } else {
+ argumentTypes.add(getTSDataType(type));
+ }
+ }
+
+ String functionName = aggregation.getResolvedFunction().getSignature().getName();
+ GroupedAccumulator accumulator =
+ createGroupedAccumulator(
+ functionName,
+ getAggregationTypeByFuncName(functionName),
+ argumentTypes,
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ true);
+
+ return new GroupedAggregator(
+ accumulator,
+ step,
+ getTSDataType(typeProvider.getTableModelType(symbol)),
+ argumentChannels,
+ OptionalInt.empty());
+ }
+
@Override
public Operator visitAggregationTableScan(
AggregationTableScanNode node, LocalExecutionPlanContext context) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
index f2909ee6bb7..c1ffa2ead26 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
@@ -315,7 +315,10 @@ public class AggregationTableScanNode extends TableScanNode {
return false;
}
AggregationTableScanNode that = (AggregationTableScanNode) o;
- return Objects.equals(projection, that.projection)
+ return Objects.equals(qualifiedObjectName, that.qualifiedObjectName)
+ && Objects.equals(outputSymbols, that.outputSymbols)
+ && Objects.equals(regionReplicaSet, that.regionReplicaSet)
+ && Objects.equals(projection, that.projection)
&& Objects.equals(aggregations, that.aggregations)
&& Objects.equals(groupingSets, that.groupingSets)
&& Objects.equals(step, that.step)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java
index 21bbcd4a811..9f8d4919ee8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/utils/TypeUtil.java
@@ -18,6 +18,12 @@
*/
package org.apache.iotdb.db.queryengine.plan.relational.utils;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.FlatHashStrategy;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.HashStrategy;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.XxHash64;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.read.common.type.BinaryType;
import org.apache.tsfile.read.common.type.BlobType;
import org.apache.tsfile.read.common.type.BooleanType;
@@ -30,17 +36,31 @@ import org.apache.tsfile.read.common.type.TimestampType;
import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.read.common.type.TypeEnum;
import org.apache.tsfile.read.common.type.UnknownType;
+import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static java.lang.Double.doubleToLongBits;
+import static java.lang.Float.floatToIntBits;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.XxHash64.FALSE_XX_HASH;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.hash.XxHash64.TRUE_XX_HASH;
import static org.apache.tsfile.read.common.type.IntType.INT32;
import static org.apache.tsfile.read.common.type.LongType.INT64;
+import static org.apache.tsfile.utils.BytesUtils.bytesToDouble;
+import static org.apache.tsfile.utils.BytesUtils.bytesToFloat;
+import static org.apache.tsfile.utils.BytesUtils.bytesToInt;
+import static org.apache.tsfile.utils.BytesUtils.bytesToLongFromOffset;
+import static org.apache.tsfile.utils.BytesUtils.doubleToBytes;
+import static org.apache.tsfile.utils.BytesUtils.floatToBytes;
+import static org.apache.tsfile.utils.BytesUtils.intToBytes;
+import static org.apache.tsfile.utils.BytesUtils.longToBytes;
public class TypeUtil {
@@ -102,4 +122,381 @@ public class TypeUtil {
return UnknownType.UNKNOWN;
}
}
+
+ // TODO move these methods into each Type to avoid branch miss
+ public static boolean isFlatVariableWidth(Type type) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case TIMESTAMP:
+ case DATE:
+ return false;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return true;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static int getFlatFixedSize(Type type) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return 1;
+ case INT32:
+ case DATE:
+ return Integer.BYTES;
+ case INT64:
+ case TIMESTAMP:
+ return Long.BYTES;
+ case FLOAT:
+ return Float.BYTES;
+ case DOUBLE:
+ return Double.BYTES;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return 16;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static int getFlatVariableWidthSize(Type type, Column column, int position) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case TIMESTAMP:
+ case DATE:
+ return 0;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return column.isNull(position) ? 0 : column.getBinary(position).getLength();
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static int getFlatVariableWidthSize(Type type, Column column, int[] position) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case TIMESTAMP:
+ case DATE:
+ return 0;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ int result = 0;
+ for (int i = 0; i < position.length; i++) {
+ if (!column.isNull(i)) {
+ result += column.getBinary(position[i]).getLength();
+ }
+ }
+ return result;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static void readFlat(
+ Type type,
+ byte[] fixedChunk,
+ int fixedOffset,
+ byte[] variableChunk,
+ ColumnBuilder columnBuilder) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ columnBuilder.writeBoolean(fixedChunk[fixedOffset] != 0);
+ break;
+ case INT32:
+ case DATE:
+ columnBuilder.writeInt(bytesToInt(fixedChunk, fixedOffset));
+ break;
+ case INT64:
+ case TIMESTAMP:
+ columnBuilder.writeLong(bytesToLongFromOffset(fixedChunk, Long.BYTES, fixedOffset));
+ break;
+ case FLOAT:
+ columnBuilder.writeFloat(bytesToFloat(fixedChunk, fixedOffset));
+ break;
+ case DOUBLE:
+ columnBuilder.writeDouble(bytesToDouble(fixedChunk, fixedOffset));
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ int length = bytesToInt(fixedChunk, fixedOffset);
+ byte[] result = new byte[length];
+ if (length <= 12) {
+ System.arraycopy(fixedChunk, fixedOffset + Integer.BYTES, result, 0, length);
+ columnBuilder.writeBinary(new Binary(result));
+ } else {
+ int variableSizeOffset = bytesToInt(fixedChunk, fixedOffset + Integer.BYTES + Long.BYTES);
+ System.arraycopy(variableChunk, variableSizeOffset, result, 0, length);
+ columnBuilder.writeBinary(new Binary(result));
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static void writeFlat(
+ Type type,
+ Column column,
+ int position,
+ byte[] fixedChunk,
+ int fixedOffset,
+ byte[] variableChunk,
+ int variableOffset) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ fixedChunk[fixedOffset] = (byte) (column.getBoolean(position) ? 1 : 0);
+ break;
+ case INT32:
+ case DATE:
+ intToBytes(column.getInt(position), fixedChunk, fixedOffset);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ longToBytes(column.getLong(position), fixedChunk, fixedOffset);
+ break;
+ case FLOAT:
+ floatToBytes(column.getFloat(position), fixedChunk, fixedOffset);
+ break;
+ case DOUBLE:
+ doubleToBytes(column.getDouble(position), fixedChunk, fixedOffset);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ byte[] value = column.getBinary(position).getValues();
+ intToBytes(value.length, fixedChunk, fixedOffset);
+ if (value.length <= 12) {
+ System.arraycopy(value, 0, fixedChunk, fixedOffset + Integer.BYTES, value.length);
+ } else {
+ intToBytes(variableOffset, fixedChunk, fixedOffset + Integer.BYTES + Long.BYTES);
+ System.arraycopy(value, 0, variableChunk, variableOffset, value.length);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static boolean notDistinctFrom(
+ Type type,
+ byte[] fixedChunk,
+ int fixedOffset,
+ byte[] variableChunk,
+ Column column,
+ int position) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return (fixedChunk[fixedOffset] != 0) == column.getBoolean(position);
+ case INT32:
+ case DATE:
+ return bytesToInt(fixedChunk, fixedOffset) == column.getInt(position);
+ case INT64:
+ case TIMESTAMP:
+ return bytesToLongFromOffset(fixedChunk, Long.BYTES, fixedOffset)
+ == column.getLong(position);
+ case FLOAT:
+ return bytesToFloat(fixedChunk, fixedOffset) == column.getFloat(position);
+ case DOUBLE:
+ return bytesToDouble(fixedChunk, fixedOffset) == column.getDouble(position);
+ case TEXT:
+ case STRING:
+ case BLOB:
+ int leftLength = bytesToInt(fixedChunk, fixedOffset);
+ byte[] leftValue = new byte[leftLength];
+ byte[] rightValue = column.getBinary(position).getValues();
+
+ if (leftLength != rightValue.length) {
+ return false;
+ }
+
+ if (leftLength <= 12) {
+ System.arraycopy(fixedChunk, fixedOffset + Integer.BYTES, leftValue, 0, leftLength);
+ } else {
+ int variableSizeOffset = bytesToInt(fixedChunk, fixedOffset + Integer.BYTES + Long.BYTES);
+ System.arraycopy(variableChunk, variableSizeOffset, leftValue, 0, leftLength);
+ }
+ return Arrays.equals(leftValue, rightValue);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ /*public static long hash(Type type, byte[] fixedChunk, int fixedOffset, byte[] variableChunk) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return fixedChunk[fixedOffset] != 0 ? TRUE_XX_HASH : FALSE_XX_HASH;
+ case INT32:
+ case DATE:
+ return rotateLeft(((long) bytesToInt(fixedChunk, fixedOffset)) * 0xC2B2AE3D27D4EB4FL, 31)
+ * 0x9E3779B185EBCA87L;
+ case INT64:
+ case TIMESTAMP:
+ // xxHash64 mix
+ return rotateLeft(
+ bytesToLongFromOffset(fixedChunk, Long.BYTES, fixedOffset) * 0xC2B2AE3D27D4EB4FL,
+ 31)
+ * 0x9E3779B185EBCA87L;
+ case FLOAT:
+ float value = bytesToFloat(fixedChunk, fixedOffset);
+ if (value == 0) {
+ return 0;
+ }
+ return rotateLeft(((long) floatToIntBits(value) * 0xC2B2AE3D27D4EB4FL), 31)
+ * 0x9E3779B185EBCA87L;
+ case DOUBLE:
+ double value1 = bytesToDouble(fixedChunk, fixedOffset);
+ if (value1 == 0) {
+ return 0;
+ }
+ return rotateLeft((doubleToLongBits(value1) * 0xC2B2AE3D27D4EB4FL), 31)
+ * 0x9E3779B185EBCA87L;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ int length = bytesToInt(fixedChunk, fixedOffset);
+ byte[] values = new byte[length];
+
+ if (length <= 12) {
+ System.arraycopy(fixedChunk, fixedOffset + Integer.BYTES, values, 0, length);
+ } else {
+ int variableSizeOffset = bytesToInt(fixedChunk, fixedOffset + Integer.BYTES + Long.BYTES);
+ System.arraycopy(variableChunk, variableSizeOffset, values, 0, length);
+ }
+ return XxHash64.hash(values);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }*/
+
+ public static long hash(Type type, byte[] fixedChunk, int fixedOffset, byte[] variableChunk) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return fixedChunk[fixedOffset] != 0 ? TRUE_XX_HASH : FALSE_XX_HASH;
+ case INT32:
+ case DATE:
+ return XxHash64.hash(bytesToInt(fixedChunk, fixedOffset));
+ case INT64:
+ case TIMESTAMP:
+ // xxHash64 mix
+
+ return XxHash64.hash(bytesToLongFromOffset(fixedChunk, Long.BYTES, fixedOffset));
+ case FLOAT:
+ float value = bytesToFloat(fixedChunk, fixedOffset);
+ if (value == 0) {
+ return 0;
+ }
+ return XxHash64.hash(floatToIntBits(value));
+ case DOUBLE:
+ double value1 = bytesToDouble(fixedChunk, fixedOffset);
+ if (value1 == 0) {
+ return 0;
+ }
+ return XxHash64.hash(doubleToLongBits(value1));
+ case TEXT:
+ case STRING:
+ case BLOB:
+ int length = bytesToInt(fixedChunk, fixedOffset);
+ byte[] values = new byte[length];
+
+ if (length <= 12) {
+ System.arraycopy(fixedChunk, fixedOffset + Integer.BYTES, values, 0, length);
+ } else {
+ int variableSizeOffset = bytesToInt(fixedChunk, fixedOffset + Integer.BYTES + Long.BYTES);
+ System.arraycopy(variableChunk, variableSizeOffset, values, 0, length);
+ }
+ return XxHash64.hash(values);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ /*public static long hash(Type type, Column column, int position) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return column.getBoolean(position) ? TRUE_XX_HASH : FALSE_XX_HASH;
+ case INT32:
+ case DATE:
+ return rotateLeft(((long) column.getInt(position)) * 0xC2B2AE3D27D4EB4FL, 31)
+ * 0x9E3779B185EBCA87L;
+ case INT64:
+ case TIMESTAMP:
+ // xxHash64 mix
+ return rotateLeft(column.getLong(position) * 0xC2B2AE3D27D4EB4FL, 31) * 0x9E3779B185EBCA87L;
+ case FLOAT:
+ float value = column.getFloat(position);
+ if (value == 0) {
+ return 0;
+ }
+ return rotateLeft(((long) floatToIntBits(value) * 0xC2B2AE3D27D4EB4FL), 31)
+ * 0x9E3779B185EBCA87L;
+ case DOUBLE:
+ double value1 = column.getDouble(position);
+ if (value1 == 0) {
+ return 0;
+ }
+ return rotateLeft((doubleToLongBits(value1) * 0xC2B2AE3D27D4EB4FL), 31)
+ * 0x9E3779B185EBCA87L;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return XxHash64.hash(column.getBinary(position).getValues());
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }*/
+
+ public static long hash(Type type, Column column, int position) {
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return column.getBoolean(position) ? TRUE_XX_HASH : FALSE_XX_HASH;
+ case INT32:
+ case DATE:
+ return XxHash64.hash(column.getInt(position));
+ case INT64:
+ case TIMESTAMP:
+ // xxHash64 mix
+ return XxHash64.hash(column.getLong(position));
+ case FLOAT:
+ float value = column.getFloat(position);
+ if (value == 0) {
+ return 0;
+ }
+ return XxHash64.hash(floatToIntBits(column.getFloat(position)));
+ case DOUBLE:
+ double value1 = column.getDouble(position);
+ if (value1 == 0) {
+ return 0;
+ }
+ return XxHash64.hash(doubleToLongBits(column.getDouble(position)));
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return XxHash64.hash(column.getBinary(position).getValues());
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public static FlatHashStrategy getFlatHashStrategy(List<Type> hashTypes) {
+ return new HashStrategy(hashTypes);
+ }
}