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/11/07 23:44:02 UTC
(iotdb) branch master updated: Support stream optimize for HashAggregation of 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 26a0da957d2 Support stream optimize for HashAggregation of TableModel
26a0da957d2 is described below
commit 26a0da957d23cf139aea297a2465c672524b0270
Author: Weihao Li <60...@users.noreply.github.com>
AuthorDate: Fri Nov 8 07:43:56 2024 +0800
Support stream optimize for HashAggregation of TableModel
---
.../relational/aggregation/AvgAccumulator.java | 4 +-
.../relational/aggregation/FirstAccumulator.java | 6 +-
.../relational/aggregation/FirstByAccumulator.java | 4 +-
.../relational/aggregation/LastAccumulator.java | 4 +-
.../relational/aggregation/LastByAccumulator.java | 4 +-
.../aggregation/TableMaxMinByBaseAccumulator.java | 4 +-
.../aggregation/TableModeAccumulator.java | 14 +
.../aggregation/TableVarianceAccumulator.java | 15 +
.../aggregation/grouped/GroupedAccumulator.java | 2 +
.../aggregation/grouped/GroupedAggregator.java | 4 +
.../aggregation/grouped/GroupedAvgAccumulator.java | 11 +-
.../grouped/GroupedCountAccumulator.java | 5 +
.../grouped/GroupedExtremeAccumulator.java | 28 ++
.../grouped/GroupedFirstAccumulator.java | 40 ++-
.../grouped/GroupedFirstByAccumulator.java | 39 ++-
.../grouped/GroupedLastAccumulator.java | 40 ++-
.../grouped/GroupedLastByAccumulator.java | 37 ++-
.../aggregation/grouped/GroupedMaxAccumulator.java | 32 ++
.../grouped/GroupedMaxMinByBaseAccumulator.java | 74 +++++
.../aggregation/grouped/GroupedMinAccumulator.java | 32 ++
.../grouped/GroupedModeAccumulator.java | 20 ++
.../aggregation/grouped/GroupedSumAccumulator.java | 6 +
.../grouped/GroupedVarianceAccumulator.java | 22 ++
.../grouped/HashAggregationOperator.java | 11 +-
.../grouped/StreamingAggregationOperator.java | 268 +++++++++++++++++
.../grouped/StreamingHashAggregationOperator.java | 329 +++++++++++++++++++++
.../aggregation/grouped/array/BinaryBigArray.java | 4 +
.../aggregation/grouped/array/BooleanBigArray.java | 4 +
.../aggregation/grouped/array/ByteBigArray.java | 4 +
.../aggregation/grouped/array/DoubleBigArray.java | 4 +
.../aggregation/grouped/array/FloatBigArray.java | 4 +
.../aggregation/grouped/array/IntBigArray.java | 4 +
.../aggregation/grouped/array/LongBigArray.java | 4 +
.../aggregation/grouped/array/MapBigArray.java | 9 +
.../aggregation/grouped/array/ObjectBigArray.java | 16 +
.../aggregation/grouped/array/ShortBigArray.java | 4 +
.../grouped/builder/HashAggregationBuilder.java | 4 +
.../builder/InMemoryHashAggregationBuilder.java | 40 ++-
.../plan/planner/TableOperatorGenerator.java | 91 +++++-
.../plan/relational/analyzer/Analysis.java | 4 +
.../distribute/TableDistributedPlanGenerator.java | 14 +-
.../PushAggregationIntoTableScan.java | 2 +-
.../TransformAggregationToStreamable.java | 2 +-
.../plan/relational/utils/TypeUtil.java | 8 +-
.../plan/relational/analyzer/AggregationTest.java | 5 +-
45 files changed, 1229 insertions(+), 53 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
index f0c6ae25af0..cfa267eac03 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java
@@ -86,7 +86,9 @@ public class AvgAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of Avg should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java
index 6a827634da9..2cdd9b8edfc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java
@@ -94,7 +94,9 @@ public class FirstAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of First should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
@@ -148,7 +150,7 @@ public class FirstAccumulator implements TableAccumulator {
public void evaluateIntermediate(ColumnBuilder columnBuilder) {
checkArgument(
columnBuilder instanceof BinaryColumnBuilder,
- "intermediate input and output of Avg should be BinaryColumn");
+ "intermediate input and output of First should be BinaryColumn");
if (!initResult) {
columnBuilder.appendNull();
} else {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstByAccumulator.java
index a829183ddfe..44eb6553c70 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstByAccumulator.java
@@ -112,7 +112,9 @@ public class FirstByAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of FirstBy should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
index a4b76759d10..34eee287ad6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastAccumulator.java
@@ -94,7 +94,9 @@ public class LastAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of Last should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
index 9a3d68c1a06..f03a3ab6f77 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/LastByAccumulator.java
@@ -112,7 +112,9 @@ public class LastByAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of LastBy should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableMaxMinByBaseAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableMaxMinByBaseAccumulator.java
index 77c4d21d095..0a94bca98f1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableMaxMinByBaseAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableMaxMinByBaseAccumulator.java
@@ -97,7 +97,9 @@ public abstract class TableMaxMinByBaseAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of max_by/min_by should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
index 1ad01095fac..70be25f0cf4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableModeAccumulator.java
@@ -25,6 +25,9 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -32,6 +35,7 @@ import org.apache.tsfile.utils.RamUsageEstimator;
import java.util.HashMap;
import java.util.Map;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.UNSUPPORTED_TYPE_MESSAGE;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.serializeBinaryValue;
import static org.apache.tsfile.utils.BytesUtils.bytesToBool;
@@ -128,6 +132,12 @@ public class TableModeAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of Mode should be BinaryColumn");
+
for (int i = 0; i < argument.getPositionCount(); i++) {
if (argument.isNull(i)) {
continue;
@@ -140,6 +150,10 @@ public class TableModeAccumulator implements TableAccumulator {
@Override
public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output should be BinaryColumn");
+
columnBuilder.writeBinary(new Binary(serializeCountMap()));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
index 0eb60b15991..3786d99f1e0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/TableVarianceAccumulator.java
@@ -25,11 +25,16 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
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 static com.google.common.base.Preconditions.checkArgument;
+
public class TableVarianceAccumulator implements TableAccumulator {
private static final long INSTANCE_SIZE =
@@ -86,6 +91,12 @@ public class TableVarianceAccumulator implements TableAccumulator {
@Override
public void addIntermediate(Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output should be BinaryColumn");
+
for (int i = 0; i < argument.getPositionCount(); i++) {
if (argument.isNull(i)) {
continue;
@@ -108,6 +119,10 @@ public class TableVarianceAccumulator implements TableAccumulator {
@Override
public void evaluateIntermediate(ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output should be BinaryColumn");
+
if (count == 0) {
columnBuilder.appendNull();
} else {
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
index f417af0dfcd..3c6ced2701e 100644
--- 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
@@ -31,4 +31,6 @@ public interface GroupedAccumulator {
void evaluateFinal(int groupId, ColumnBuilder columnBuilder);
void prepareFinal();
+
+ void reset();
}
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
index 4e73193aab9..4e1b8c875f0 100644
--- 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
@@ -85,6 +85,10 @@ public class GroupedAggregator {
accumulator.prepareFinal();
}
+ public void reset() {
+ accumulator.reset();
+ }
+
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
index 1fa67755c2a..a1ef5f71b5d 100644
--- 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
@@ -26,6 +26,7 @@ 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.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -88,7 +89,9 @@ public class GroupedAvgAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of Avg should be BinaryColumn");
for (int i = 0; i < groupIds.length; i++) {
@@ -174,4 +177,10 @@ public class GroupedAvgAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+
+ @Override
+ public void reset() {
+ countValues.reset();
+ sumValues.reset();
+ }
}
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
index 2d75b7b824b..992d9a5b050 100644
--- 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
@@ -70,4 +70,9 @@ public class GroupedCountAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+
+ @Override
+ public void reset() {
+ countValues.reset();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedExtremeAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedExtremeAccumulator.java
index 8ee8169ab77..1cca69c9211 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedExtremeAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedExtremeAccumulator.java
@@ -256,6 +256,34 @@ public class GroupedExtremeAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ inits.reset();
+ switch (seriesDataType) {
+ case INT32:
+ intValues.reset();
+ return;
+ case INT64:
+ longValues.reset();
+ return;
+ case FLOAT:
+ floatValues.reset();
+ return;
+ case DOUBLE:
+ doubleValues.reset();
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ case BOOLEAN:
+ case DATE:
+ case TIMESTAMP:
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", seriesDataType));
+ }
+ }
+
private void addIntInput(int[] groupIds, Column valueColumn) {
for (int i = 0; i < groupIds.length; i++) {
if (!valueColumn.isNull(i)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
index 94962905d2c..1db66df8190 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstAccumulator.java
@@ -188,8 +188,10 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
- "intermediate input and output should be BinaryColumn");
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of First should be BinaryColumn");
for (int i = 0; i < groupIds.length; i++) {
if (argument.isNull(i)) {
@@ -242,7 +244,7 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
checkArgument(
columnBuilder instanceof BinaryColumnBuilder,
- "intermediate input and output of should be BinaryColumn");
+ "intermediate input and output of First should be BinaryColumn");
if (minTimes.get(groupId) == Long.MIN_VALUE) {
columnBuilder.appendNull();
} else {
@@ -288,6 +290,38 @@ public class GroupedFirstAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ minTimes.reset();
+ switch (seriesDataType) {
+ case INT32:
+ case DATE:
+ intValues.reset();
+ return;
+ case INT64:
+ case TIMESTAMP:
+ longValues.reset();
+ return;
+ case FLOAT:
+ floatValues.reset();
+ return;
+ case DOUBLE:
+ doubleValues.reset();
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ binaryValues.reset();
+ return;
+ case BOOLEAN:
+ booleanValues.reset();
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", seriesDataType));
+ }
+ }
+
private byte[] serializeTimeWithValue(int groupId) {
byte[] bytes;
int length = Long.BYTES;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
index d23210fdc20..1da754819cf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedFirstByAccumulator.java
@@ -168,6 +168,39 @@ public class GroupedFirstByAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ yFirstTimes.reset();
+ inits.reset();
+ xNulls.reset();
+ switch (xDataType) {
+ case INT32:
+ case DATE:
+ xIntValues.reset();
+ break;
+ case INT64:
+ case TIMESTAMP:
+ xLongValues.reset();
+ break;
+ case FLOAT:
+ xFloatValues.reset();
+ break;
+ case DOUBLE:
+ xDoubleValues.reset();
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ xBinaryValues.reset();
+ break;
+ case BOOLEAN:
+ xBooleanValues.reset();
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", xDataType));
+ }
+ }
+
@Override
public void addInput(int[] groupIds, Column[] arguments) {
checkArgument(arguments.length == 3, "Length of input Column[] for LastBy/FirstBy should be 3");
@@ -205,8 +238,10 @@ public class GroupedFirstByAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
- "intermediate input and output of LastBy should be BinaryColumn");
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of FirstBy should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
if (argument.isNull(i)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
index c230b8aa9e7..6175bd0428d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastAccumulator.java
@@ -188,8 +188,10 @@ public class GroupedLastAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
- "intermediate input and output should be BinaryColumn");
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of Last should be BinaryColumn");
for (int i = 0; i < groupIds.length; i++) {
if (argument.isNull(i)) {
@@ -242,7 +244,7 @@ public class GroupedLastAccumulator implements GroupedAccumulator {
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
checkArgument(
columnBuilder instanceof BinaryColumnBuilder,
- "intermediate input and output of should be BinaryColumn");
+ "intermediate input and output of Last should be BinaryColumn");
if (maxTimes.get(groupId) == Long.MIN_VALUE) {
columnBuilder.appendNull();
} else {
@@ -288,6 +290,38 @@ public class GroupedLastAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ maxTimes.reset();
+ switch (seriesDataType) {
+ case INT32:
+ case DATE:
+ intValues.reset();
+ return;
+ case INT64:
+ case TIMESTAMP:
+ longValues.reset();
+ return;
+ case FLOAT:
+ floatValues.reset();
+ return;
+ case DOUBLE:
+ doubleValues.reset();
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ binaryValues.reset();
+ return;
+ case BOOLEAN:
+ booleanValues.reset();
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", seriesDataType));
+ }
+ }
+
private byte[] serializeTimeWithValue(int groupId) {
byte[] bytes;
int length = Long.BYTES;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
index 77b2ec762d6..f93a20f66e3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedLastByAccumulator.java
@@ -169,6 +169,39 @@ public class GroupedLastByAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ yLastTimes.reset();
+ inits.reset();
+ xNulls.reset();
+ switch (xDataType) {
+ case INT32:
+ case DATE:
+ xIntValues.reset();
+ break;
+ case INT64:
+ case TIMESTAMP:
+ xLongValues.reset();
+ break;
+ case FLOAT:
+ xFloatValues.reset();
+ break;
+ case DOUBLE:
+ xDoubleValues.reset();
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ xBinaryValues.reset();
+ break;
+ case BOOLEAN:
+ xBooleanValues.reset();
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", xDataType));
+ }
+ }
+
@Override
public void addInput(int[] groupIds, Column[] arguments) {
checkArgument(arguments.length == 3, "Length of input Column[] for LastBy/LastBy should be 3");
@@ -206,7 +239,9 @@ public class GroupedLastByAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
checkArgument(
- argument instanceof BinaryColumn || argument instanceof RunLengthEncodedColumn,
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
"intermediate input and output of LastBy should be BinaryColumn");
for (int i = 0; i < argument.getPositionCount(); i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxAccumulator.java
index 4d954a99591..fc907d34a40 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxAccumulator.java
@@ -289,6 +289,38 @@ public class GroupedMaxAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ inits.reset();
+ switch (seriesDataType) {
+ case INT32:
+ case DATE:
+ intValues.reset();
+ return;
+ case INT64:
+ case TIMESTAMP:
+ longValues.reset();
+ return;
+ case FLOAT:
+ floatValues.reset();
+ return;
+ case DOUBLE:
+ doubleValues.reset();
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ binaryValues.reset();
+ return;
+ case BOOLEAN:
+ booleanValues.reset();
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", seriesDataType));
+ }
+ }
+
private void addIntInput(int[] groupIds, Column valueColumn) {
for (int i = 0; i < groupIds.length; i++) {
if (!valueColumn.isNull(i)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxMinByBaseAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxMinByBaseAccumulator.java
index 230bd170664..bdee50b9a15 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxMinByBaseAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMaxMinByBaseAccumulator.java
@@ -30,6 +30,9 @@ 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.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -37,6 +40,7 @@ import org.apache.tsfile.write.UnSupportedDataTypeException;
import java.util.Collections;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.tsfile.utils.BytesUtils.boolToBytes;
import static org.apache.tsfile.utils.BytesUtils.doubleToBytes;
import static org.apache.tsfile.utils.BytesUtils.floatToBytes;
@@ -256,6 +260,67 @@ public abstract class GroupedMaxMinByBaseAccumulator implements GroupedAccumulat
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ inits.reset();
+ xNulls.reset();
+ switch (xDataType) {
+ case INT32:
+ case DATE:
+ xIntValues.reset();
+ break;
+ case INT64:
+ case TIMESTAMP:
+ xLongValues.reset();
+ break;
+ case FLOAT:
+ xFloatValues.reset();
+ break;
+ case DOUBLE:
+ xDoubleValues.reset();
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ xBinaryValues.reset();
+ break;
+ case BOOLEAN:
+ xBooleanValues.reset();
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", xDataType));
+ }
+
+ switch (yDataType) {
+ case INT32:
+ case DATE:
+ yIntValues.reset();
+ break;
+ case INT64:
+ case TIMESTAMP:
+ yLongValues.reset();
+ break;
+ case FLOAT:
+ yFloatValues.reset();
+ break;
+ case DOUBLE:
+ yDoubleValues.reset();
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ yBinaryValues.reset();
+ break;
+ case BOOLEAN:
+ yBooleanValues.reset();
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", yDataType));
+ }
+ }
+
@Override
public void addInput(int[] groupIds, Column[] arguments) {
switch (yDataType) {
@@ -289,6 +354,11 @@ public abstract class GroupedMaxMinByBaseAccumulator implements GroupedAccumulat
@Override
public void addIntermediate(int[] groupIds, Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of MaxBy or MinBy should be BinaryColumn");
for (int i = 0; i < groupIds.length; i++) {
if (argument.isNull(i)) {
@@ -302,6 +372,10 @@ public abstract class GroupedMaxMinByBaseAccumulator implements GroupedAccumulat
@Override
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output of Mode should be BinaryColumn");
+
if (!inits.get(groupId)) {
columnBuilder.appendNull();
return;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMinAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMinAccumulator.java
index a86fa942717..c38dc254470 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMinAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedMinAccumulator.java
@@ -289,6 +289,38 @@ public class GroupedMinAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ inits.reset();
+ switch (seriesDataType) {
+ case INT32:
+ case DATE:
+ intValues.reset();
+ return;
+ case INT64:
+ case TIMESTAMP:
+ longValues.reset();
+ return;
+ case FLOAT:
+ floatValues.reset();
+ return;
+ case DOUBLE:
+ doubleValues.reset();
+ return;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ binaryValues.reset();
+ return;
+ case BOOLEAN:
+ booleanValues.reset();
+ return;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Unsupported data type : %s", seriesDataType));
+ }
+ }
+
private void addIntInput(int[] groupIds, Column valueColumn) {
for (int i = 0; i < groupIds.length; i++) {
if (!valueColumn.isNull(i)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedModeAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedModeAccumulator.java
index f3a5a316760..eddf3ce945a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedModeAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedModeAccumulator.java
@@ -26,6 +26,9 @@ import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggr
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.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -34,6 +37,7 @@ import org.apache.tsfile.utils.TsPrimitiveType;
import java.util.HashMap;
import java.util.Map;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.UNSUPPORTED_TYPE_MESSAGE;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.serializeBinaryValue;
import static org.apache.tsfile.utils.BytesUtils.bytesToBool;
@@ -100,6 +104,12 @@ public class GroupedModeAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output of Mode should be BinaryColumn");
+
for (int i = 0; i < argument.getPositionCount(); i++) {
if (argument.isNull(i)) {
continue;
@@ -112,6 +122,10 @@ public class GroupedModeAccumulator implements GroupedAccumulator {
@Override
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output of Mode should be BinaryColumn");
+
columnBuilder.writeBinary(new Binary(serializeCountMap(groupId)));
}
@@ -162,6 +176,12 @@ public class GroupedModeAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ countMaps.reset();
+ nullCounts.reset();
+ }
+
// haveNull | nullCount (optional) | countMap
private byte[] serializeCountMap(int groupId) {
byte[] bytes;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedSumAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedSumAccumulator.java
index 0ba6d859de2..67bd9c273a0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedSumAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedSumAccumulator.java
@@ -149,4 +149,10 @@ public class GroupedSumAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+
+ @Override
+ public void reset() {
+ initResult.reset();
+ sumValues.reset();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
index c80406e93c4..64d230734d0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/GroupedVarianceAccumulator.java
@@ -26,11 +26,16 @@ import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggr
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.read.common.block.column.RunLengthEncodedColumn;
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 static com.google.common.base.Preconditions.checkArgument;
+
public class GroupedVarianceAccumulator implements GroupedAccumulator {
private static final long INSTANCE_SIZE =
@@ -89,6 +94,12 @@ public class GroupedVarianceAccumulator implements GroupedAccumulator {
@Override
public void addIntermediate(int[] groupIds, Column argument) {
+ checkArgument(
+ argument instanceof BinaryColumn
+ || (argument instanceof RunLengthEncodedColumn
+ && ((RunLengthEncodedColumn) argument).getValue() instanceof BinaryColumn),
+ "intermediate input and output should be BinaryColumn");
+
for (int i = 0; i < argument.getPositionCount(); i++) {
if (argument.isNull(i)) {
continue;
@@ -116,6 +127,10 @@ public class GroupedVarianceAccumulator implements GroupedAccumulator {
@Override
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
+ checkArgument(
+ columnBuilder instanceof BinaryColumnBuilder,
+ "intermediate input and output should be BinaryColumn");
+
if (counts.get(groupId) == 0) {
columnBuilder.appendNull();
} else {
@@ -167,6 +182,13 @@ public class GroupedVarianceAccumulator implements GroupedAccumulator {
@Override
public void prepareFinal() {}
+ @Override
+ public void reset() {
+ counts.reset();
+ means.reset();
+ m2s.reset();
+ }
+
private void addIntInput(int[] groupIds, Column column) {
for (int i = 0; i < column.getPositionCount(); i++) {
if (column.isNull(i)) {
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
index a2fca3c829b..bbeb468088b 100644
--- 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
@@ -165,15 +165,14 @@ public class HashAggregationOperator extends AbstractOperator {
}
private TsBlock getOutput() {
- // only flush if we are finishing or the aggregation builder is full
- if (aggregationBuilder == null) {
- return null;
- }
+ checkState(aggregationBuilder != null);
TsBlock result = aggregationBuilder.buildResult();
- closeAggregationBuilder();
- finished = true;
+ if (aggregationBuilder.finished()) {
+ closeAggregationBuilder();
+ finished = true;
+ }
return result;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java
new file mode 100644
index 00000000000..5b89d16232f
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingAggregationOperator.java
@@ -0,0 +1,268 @@
+/*
+ * 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.TableScanOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
+import org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+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 org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class StreamingAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(StreamingAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final List<TableAggregator> aggregators;
+
+ private final int[] groupByChannels;
+
+ private final TsBlockBuilder resultBuilder;
+ private final ColumnBuilder[] resultColumnsBuilder;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality
+ private SortKey currentGroup;
+ private final Comparator<SortKey> groupKeyComparator;
+
+ // more than one group in input block
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Type> groupByTypes,
+ List<Integer> groupByChannels,
+ Comparator<SortKey> groupKeyComparator,
+ List<TableAggregator> aggregators,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+ this.groupByChannels = Ints.toArray(groupByChannels);
+ this.groupKeyComparator = groupKeyComparator;
+ this.aggregators = aggregators;
+ this.resultBuilder =
+ new TsBlockBuilder(
+ Stream.concat(
+ groupByTypes.stream().map(InternalTypeManager::getTSDataType),
+ aggregators.stream().map(TableAggregator::getType))
+ .collect(Collectors.toList()));
+ this.resultColumnsBuilder = resultBuilder.getValueColumnBuilders();
+ checkArgument(!spillEnabled, "spill is not supported");
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup.tsBlock, currentGroup.rowIndex, true);
+ currentGroup = null;
+ }
+ finished = true;
+ }
+ checkState(!outputs.isEmpty(), "outputs should always not be empty here");
+
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ private void processInput(TsBlock page) {
+ requireNonNull(page, "page is null");
+
+ if (currentGroup != null) {
+ if (groupKeyComparator.compare(currentGroup, new SortKey(page, 0)) != 0) {
+ // page starts with new group, so flush it
+ evaluateAndFlushGroup(currentGroup.tsBlock, currentGroup.rowIndex, false);
+ }
+ currentGroup = null;
+ }
+
+ int startPosition = 0;
+ while (true) {
+ // may be equal to page.getPositionCount() if the end is not found in this page
+ int nextGroupStart = findNextGroupStart(startPosition, page);
+ addRowsToAggregators(page, startPosition, nextGroupStart - 1);
+
+ if (nextGroupStart < page.getPositionCount()) {
+ // current group stops somewhere in the middle of the page, so flush it
+ evaluateAndFlushGroup(page, startPosition, false);
+ startPosition = nextGroupStart;
+ } else {
+ // late materialization requires that page being locally stored is materialized before the
+ // next one is fetched
+ currentGroup = new SortKey(page, page.getPositionCount() - 1);
+ return;
+ }
+ }
+ }
+
+ private void addRowsToAggregators(TsBlock page, int startPosition, int endPosition) {
+ TsBlock region = page.getRegion(startPosition, endPosition - startPosition + 1);
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.processBlock(region);
+ }
+ }
+
+ private void resetAggregationBuilder() {
+ for (TableAggregator aggregator : aggregators) {
+ aggregator.reset();
+ }
+ }
+
+ private void evaluateAndFlushGroup(TsBlock page, int position, boolean lastCalculate) {
+ resultBuilder.declarePosition();
+ for (int i = 0; i < groupByChannels.length; i++) {
+ Column input = page.getColumn(groupByChannels[i]);
+ if (input.isNull(position)) {
+ resultColumnsBuilder[i].appendNull();
+ } else {
+ resultColumnsBuilder[i].write(input, position);
+ }
+ }
+ int offset = groupByChannels.length;
+ for (int i = 0; i < aggregators.size(); i++) {
+ aggregators.get(i).evaluate(resultColumnsBuilder[offset + i]);
+ }
+
+ if (lastCalculate || resultBuilder.isFull()) {
+ outputs.add(
+ resultBuilder.build(
+ new RunLengthEncodedColumn(
+ TableScanOperator.TIME_COLUMN_TEMPLATE, resultBuilder.getPositionCount())));
+ resultBuilder.reset();
+ }
+
+ resetAggregationBuilder();
+ }
+
+ private int findNextGroupStart(int startPosition, TsBlock page) {
+ int positionCount = page.getPositionCount();
+ for (int i = startPosition + 1; i < positionCount; i++) {
+ if (groupKeyComparator.compare(new SortKey(page, startPosition), new SortKey(page, i)) != 0) {
+ return i;
+ }
+ }
+
+ return positionCount;
+ }
+
+ @Override
+ public boolean isFinished() throws Exception {
+ return finished && retainedTsBlock == null && outputs.isEmpty();
+ }
+
+ @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)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ + outputs.stream().mapToLong(TsBlock::getRetainedSizeInBytes).sum()
+ + resultBuilder.getRetainedSizeInBytes();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingHashAggregationOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingHashAggregationOperator.java
new file mode 100644
index 00000000000..dc2151b0304
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/StreamingHashAggregationOperator.java
@@ -0,0 +1,329 @@
+/*
+ * 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 org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory.NOOP;
+
+public class StreamingHashAggregationOperator extends AbstractOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(StreamingHashAggregationOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator child;
+
+ private final int[] preGroupedChannels;
+
+ // used for build result
+ private final int[] preGroupedIndexInResult;
+ private final int[] unPreGroupedIndexInResult;
+
+ private final int valueColumnsCount;
+ private final int resultColumnsCount;
+
+ private HashAggregationBuilder aggregationBuilder;
+
+ private final MemoryReservationManager memoryReservationManager;
+ // memory already occupied by aggregationBuilder
+ private long previousRetainedSize = 0;
+
+ private boolean finished = false;
+
+ // cached current group to judge row equality and construct preGroupedColumn in result
+ private SortKey currentGroup;
+ private final Comparator<SortKey> groupKeyComparator;
+
+ // more than one Blocks may be produced:
+ // 1. more than one group in input block
+ // 2. one group but the evaluated result are split
+ private final Deque<TsBlock> outputs = new LinkedList<>();
+
+ public StreamingHashAggregationOperator(
+ OperatorContext operatorContext,
+ Operator child,
+ List<Integer> preGroupedChannels,
+ List<Integer> preGroupedIndexInResult,
+ List<Type> unPreGroupedTypes,
+ List<Integer> unPreGroupedChannels,
+ List<Integer> unPreGroupedIndexInResult,
+ Comparator<SortKey> groupKeyComparator,
+ List<GroupedAggregator> aggregators,
+ AggregationNode.Step step,
+ int expectedGroups,
+ long maxPartialMemory,
+ boolean spillEnabled,
+ long unSpillMemoryLimit) {
+ this.operatorContext = operatorContext;
+ this.child = child;
+
+ this.preGroupedChannels = Ints.toArray(preGroupedChannels);
+ this.preGroupedIndexInResult = Ints.toArray(preGroupedIndexInResult);
+ this.unPreGroupedIndexInResult = Ints.toArray(unPreGroupedIndexInResult);
+
+ this.groupKeyComparator = groupKeyComparator;
+
+ this.valueColumnsCount = aggregators.size();
+ this.resultColumnsCount =
+ this.preGroupedIndexInResult.length
+ + this.unPreGroupedIndexInResult.length
+ + aggregators.size();
+ checkArgument(!spillEnabled, "spill is not supported");
+ aggregationBuilder =
+ new InMemoryHashAggregationBuilder(
+ aggregators,
+ step,
+ expectedGroups,
+ unPreGroupedTypes,
+ unPreGroupedChannels,
+ Optional.empty(),
+ operatorContext,
+ maxPartialMemory,
+ NOOP);
+ this.memoryReservationManager =
+ operatorContext
+ .getDriverContext()
+ .getFragmentInstanceContext()
+ .getMemoryReservationContext();
+ updateOccupiedMemorySize();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ return child.isBlocked();
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return !finished || retainedTsBlock != null || !outputs.isEmpty();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ if (!outputs.isEmpty()) {
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ TsBlock block;
+ if (child.hasNextWithTimer()) {
+ block = child.nextWithTimer();
+ if (block == null) {
+ return null;
+ }
+
+ processInput(block);
+
+ if (outputs.isEmpty()) {
+ return null;
+ }
+ } else {
+ // last evaluate
+ if (currentGroup != null) {
+ evaluateAndFlushGroup(currentGroup.tsBlock, currentGroup.rowIndex);
+ currentGroup = null;
+ }
+ finished = true;
+ closeAggregationBuilder();
+ }
+ checkState(!outputs.isEmpty(), "outputs should always not be empty here");
+
+ resultTsBlock = outputs.removeFirst();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ private void processInput(TsBlock page) {
+ requireNonNull(page, "page is null");
+
+ if (currentGroup != null) {
+ if (groupKeyComparator.compare(currentGroup, new SortKey(page, 0)) != 0) {
+ // page starts with new group, so flush it
+ evaluateAndFlushGroup(currentGroup.tsBlock, currentGroup.rowIndex);
+ }
+ currentGroup = null;
+ }
+
+ int startPosition = 0;
+ while (true) {
+ // may be equal to page.getPositionCount() if the end is not found in this page
+ int nextGroupStart = findNextGroupStart(startPosition, page);
+ addRowsToAggregationBuilder(page, startPosition, nextGroupStart - 1);
+
+ if (nextGroupStart < page.getPositionCount()) {
+ // current group stops somewhere in the middle of the page, so flush it
+ evaluateAndFlushGroup(page, startPosition);
+ startPosition = nextGroupStart;
+ } else {
+ // late materialization requires that page being locally stored is materialized before the
+ // next one is fetched
+ currentGroup = new SortKey(page, page.getPositionCount() - 1);
+ return;
+ }
+ }
+ }
+
+ private void addRowsToAggregationBuilder(TsBlock page, int startPosition, int endPosition) {
+ TsBlock region = page.getRegion(startPosition, endPosition - startPosition + 1);
+ aggregationBuilder.processBlock(region);
+ updateOccupiedMemorySize();
+ }
+
+ private void resetAggregationBuilder() {
+ aggregationBuilder.reset();
+ updateOccupiedMemorySize();
+ }
+
+ private void evaluateAndFlushGroup(TsBlock page, int position) {
+ Column[] result = new Column[resultColumnsCount];
+ // offset of value columns index
+ int offset = preGroupedIndexInResult.length + unPreGroupedIndexInResult.length;
+
+ do {
+ // contains unPreGrouped group by columns and value columns
+ TsBlock buildResult = aggregationBuilder.buildResult();
+
+ // preGrouped Columns
+ for (int i = 0; i < preGroupedIndexInResult.length; i++) {
+ Column column = page.getColumn(preGroupedChannels[i]).getRegion(position, 1);
+ result[preGroupedIndexInResult[i]] =
+ new RunLengthEncodedColumn(column, buildResult.getPositionCount());
+ }
+
+ // unPreGrouped Columns
+ for (int i = 0; i < unPreGroupedIndexInResult.length; i++) {
+ result[unPreGroupedIndexInResult[i]] = buildResult.getColumn(i);
+ }
+
+ // value Columns
+ for (int i = 0; i < valueColumnsCount; i++) {
+ result[offset + i] = buildResult.getColumn(i + unPreGroupedIndexInResult.length);
+ }
+
+ outputs.add(
+ TsBlock.wrapBlocksWithoutCopy(
+ buildResult.getPositionCount(), buildResult.getTimeColumn(), result));
+ } while (!aggregationBuilder.finished());
+
+ resetAggregationBuilder();
+ }
+
+ private int findNextGroupStart(int startPosition, TsBlock page) {
+ int positionCount = page.getPositionCount();
+ for (int i = startPosition + 1; i < positionCount; i++) {
+ if (groupKeyComparator.compare(new SortKey(page, startPosition), new SortKey(page, i)) != 0) {
+ return i;
+ }
+ }
+
+ return positionCount;
+ }
+
+ private void updateOccupiedMemorySize() {
+ long memorySize = aggregationBuilder.getEstimatedSize();
+ long delta = memorySize - previousRetainedSize;
+ if (delta > 0) {
+ memoryReservationManager.reserveMemoryCumulatively(delta);
+ } else if (delta < 0) {
+ memoryReservationManager.releaseMemoryCumulatively(-delta);
+ }
+ previousRetainedSize = memorySize;
+ }
+
+ 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 afterward to avoid unaccounted memory.
+ aggregationBuilder = null;
+ }
+ // memoryContext.setBytes(0);
+ }
+
+ @Override
+ public boolean isFinished() throws Exception {
+ return finished && retainedTsBlock == null && outputs.isEmpty();
+ }
+
+ @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)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ + outputs.stream().mapToLong(TsBlock::getRetainedSizeInBytes).sum();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BinaryBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BinaryBigArray.java
index 0489c8be40d..0ba101a1976 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BinaryBigArray.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/BinaryBigArray.java
@@ -73,4 +73,8 @@ public final class BinaryBigArray {
sizeOfBinarys += sizeOfObject(value);
}
}
+
+ public void reset() {
+ array.reset();
+ }
}
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
index a85d50c1f91..7e3b284b145 100644
--- 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
@@ -96,6 +96,10 @@ public final class BooleanBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
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
index fe9ce4bb2c7..fd3c0379855 100644
--- 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
@@ -92,6 +92,10 @@ public final class ByteBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
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
index 644d81e03a5..790be2458f4 100644
--- 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
@@ -105,6 +105,10 @@ public final class DoubleBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/FloatBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/FloatBigArray.java
index 03869577b26..061f2916368 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/FloatBigArray.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/FloatBigArray.java
@@ -105,6 +105,10 @@ public final class FloatBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
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
index 1ca45e897c4..56fbc3fa222 100644
--- 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
@@ -118,6 +118,10 @@ public final class IntBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
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
index 3e4a1f74f72..25db32f8aa9 100644
--- 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
@@ -159,6 +159,10 @@ public final class LongBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
private void grow(long length) {
// how many segments are required to get to the length?
int requiredSegments = segment(length) + 1;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/MapBigArray.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/MapBigArray.java
index ca674dbf2e6..1c609c12831 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/MapBigArray.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/grouped/array/MapBigArray.java
@@ -80,4 +80,13 @@ public final class MapBigArray {
sizeOfMaps += sizeOfObject(value);
}
}
+
+ public void reset() {
+ array.forEach(
+ v -> {
+ if (v != null) {
+ v.clear();
+ }
+ });
+ }
}
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
index 14690d24ca8..a2707cc7023 100644
--- 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
@@ -14,6 +14,7 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array;
import java.util.Arrays;
+import java.util.function.Consumer;
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;
@@ -113,6 +114,21 @@ public final class ObjectBigArray<T> {
}
}
+ public void reset() {
+ fill((T) initialValue);
+ }
+
+ public void forEach(Consumer<T> action) {
+ for (Object[] segment : array) {
+ if (segment == null) {
+ return;
+ }
+ for (T object : (T[]) segment) {
+ action.accept(object);
+ }
+ }
+ }
+
/**
* 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
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
index 275f654488c..1cb599e38ca 100644
--- 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
@@ -116,6 +116,10 @@ public final class ShortBigArray {
}
}
+ public void reset() {
+ fill(initialValue);
+ }
+
/**
* 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
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
index 0a420e68de1..172fe327e62 100644
--- 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
@@ -20,12 +20,16 @@ public interface HashAggregationBuilder extends AutoCloseable {
TsBlock buildResult();
+ boolean finished();
+
long getEstimatedSize();
boolean isFull();
void updateMemory();
+ void reset();
+
@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
index ad5231f939a..0fbaef5e337 100644
--- 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
@@ -45,7 +45,7 @@ import static org.apache.iotdb.db.queryengine.execution.operator.source.relation
public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
private final int[] groupByChannels;
- private final GroupByHash groupByHash;
+ private GroupByHash groupByHash;
private final List<Type> groupByOutputTypes;
private final List<GroupedAggregator> groupedAggregators;
private final boolean partial;
@@ -54,6 +54,12 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
private boolean full;
+ private Iterator<Integer> groupIds;
+ private final TsBlockBuilder pageBuilder;
+
+ private final int expectedGroups;
+ private final Optional<Integer> hashChannel;
+
public InMemoryHashAggregationBuilder(
List<GroupedAggregator> groupedAggregators,
AggregationNode.Step step,
@@ -97,6 +103,11 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
this.partial = step.isOutputPartial();
this.maxPartialMemory = OptionalLong.of(maxPartialMemory);
this.updateMemory = updateMemory;
+
+ this.pageBuilder = new TsBlockBuilder(buildTypes());
+
+ this.expectedGroups = expectedGroups;
+ this.hashChannel = hashChannel;
}
@Override
@@ -120,6 +131,18 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
// updateMemory.update();
}
+ @Override
+ public void reset() {
+ // TODO reset of GroupByHash
+ groupByHash =
+ createGroupByHash(
+ groupByOutputTypes, hashChannel.isPresent(), expectedGroups, updateMemory);
+ groupedAggregators.forEach(GroupedAggregator::reset);
+ full = false;
+ groupIds = null;
+ pageBuilder.reset();
+ }
+
@Override
public boolean isFull() {
return full;
@@ -174,6 +197,11 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
return buildResult(consecutiveGroupIds());
}
+ @Override
+ public boolean finished() {
+ return !groupIds.hasNext();
+ }
+
public List<Type> buildSpillTypes() {
ArrayList<Type> types = new ArrayList<>(groupByOutputTypes);
for (GroupedAggregator groupedAggregator : groupedAggregators) {
@@ -187,12 +215,9 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
}
private TsBlock buildResult(Iterator<Integer> groupIds) {
- TsBlockBuilder pageBuilder = new TsBlockBuilder(buildTypes());
-
pageBuilder.reset();
- // TODO Memory Control Weihao Li
- while (groupIds.hasNext()) {
+ while (!pageBuilder.isFull() && groupIds.hasNext()) {
int groupId = groupIds.next();
groupByHash.appendValuesTo(groupId, pageBuilder);
@@ -223,6 +248,9 @@ public class InMemoryHashAggregationBuilder implements HashAggregationBuilder {
}
private Iterator<Integer> consecutiveGroupIds() {
- return IntStream.range(0, groupByHash.getGroupCount()).iterator();
+ if (groupIds == null) {
+ groupIds = IntStream.range(0, groupByHash.getGroupCount()).iterator();
+ }
+ return groupIds;
}
}
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 6c2ccb750f3..d5d3327dbf9 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
@@ -78,6 +78,8 @@ import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggr
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.operator.source.relational.aggregation.grouped.StreamingAggregationOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.StreamingHashAggregationOperator;
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;
@@ -139,6 +141,7 @@ import org.apache.iotdb.db.utils.datastructure.SortKey;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
@@ -1405,16 +1408,9 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
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()
@@ -1422,9 +1418,79 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
.collect(toImmutableList());
if (node.isStreamable()) {
- /*return new StreamingAggregationOperator;*/
+ if (groupByTypes.size() == node.getPreGroupedSymbols().size()) {
+ ImmutableList.Builder<TableAggregator> aggregatorBuilder = new ImmutableList.Builder<>();
+ node.getAggregations()
+ .forEach(
+ (k, v) ->
+ aggregatorBuilder.add(
+ buildAggregator(childLayout, k, v, node.getStep(), typeProvider, true)));
+
+ return new StreamingAggregationOperator(
+ operatorContext,
+ child,
+ groupByTypes,
+ groupByChannels,
+ genGroupKeyComparator(groupByTypes, groupByChannels),
+ aggregatorBuilder.build(),
+ Long.MAX_VALUE,
+ false,
+ Long.MAX_VALUE);
+ }
+
+ ImmutableList.Builder<GroupedAggregator> aggregatorBuilder = new ImmutableList.Builder<>();
+ node.getAggregations()
+ .forEach(
+ (k, v) ->
+ aggregatorBuilder.add(
+ buildGroupByAggregator(childLayout, k, v, node.getStep(), typeProvider)));
+
+ Set<Symbol> preGroupedKeys = ImmutableSet.copyOf(node.getPreGroupedSymbols());
+ List<Symbol> groupingKeys = node.getGroupingKeys();
+ ImmutableList.Builder<Type> preGroupedTypesBuilder = new ImmutableList.Builder<>();
+ ImmutableList.Builder<Integer> preGroupedChannelsBuilder = new ImmutableList.Builder<>();
+ ImmutableList.Builder<Integer> preGroupedIndexInResultBuilder = new ImmutableList.Builder<>();
+ ImmutableList.Builder<Type> unPreGroupedTypesBuilder = new ImmutableList.Builder<>();
+ ImmutableList.Builder<Integer> unPreGroupedChannelsBuilder = new ImmutableList.Builder<>();
+ ImmutableList.Builder<Integer> unPreGroupedIndexInResultBuilder =
+ new ImmutableList.Builder<>();
+ for (int i = 0; i < groupByTypes.size(); i++) {
+ if (preGroupedKeys.contains(groupingKeys.get(i))) {
+ preGroupedTypesBuilder.add(groupByTypes.get(i));
+ preGroupedChannelsBuilder.add(groupByChannels.get(i));
+ preGroupedIndexInResultBuilder.add(i);
+ } else {
+ unPreGroupedTypesBuilder.add(groupByTypes.get(i));
+ unPreGroupedChannelsBuilder.add(groupByChannels.get(i));
+ unPreGroupedIndexInResultBuilder.add(i);
+ }
+ }
+
+ List<Integer> preGroupedChannels = preGroupedChannelsBuilder.build();
+ return new StreamingHashAggregationOperator(
+ operatorContext,
+ child,
+ preGroupedChannels,
+ preGroupedIndexInResultBuilder.build(),
+ unPreGroupedTypesBuilder.build(),
+ unPreGroupedChannelsBuilder.build(),
+ unPreGroupedIndexInResultBuilder.build(),
+ genGroupKeyComparator(preGroupedTypesBuilder.build(), preGroupedChannels),
+ aggregatorBuilder.build(),
+ node.getStep(),
+ 64,
+ Long.MAX_VALUE,
+ false,
+ Long.MAX_VALUE);
}
+ ImmutableList.Builder<GroupedAggregator> aggregatorBuilder = new ImmutableList.Builder<>();
+ node.getAggregations()
+ .forEach(
+ (k, v) ->
+ aggregatorBuilder.add(
+ buildGroupByAggregator(childLayout, k, v, node.getStep(), typeProvider)));
+
return new HashAggregationOperator(
operatorContext,
child,
@@ -1438,6 +1504,15 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Long.MAX_VALUE);
}
+ private Comparator<SortKey> genGroupKeyComparator(
+ List<Type> groupTypes, List<Integer> groupByChannels) {
+ return getComparatorForTable(
+ // SortOrder is not sensitive here, the comparator is just used to judge equality.
+ groupTypes.stream().map(k -> ASC_NULLS_LAST).collect(Collectors.toList()),
+ groupByChannels,
+ groupTypes.stream().map(InternalTypeManager::getTSDataType).collect(Collectors.toList()));
+ }
+
private static List<Integer> getChannelsForSymbols(
List<Symbol> symbols, Map<Symbol, Integer> layout) {
ImmutableList.Builder<Integer> builder = ImmutableList.builder();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
index 76b80d842eb..51351f8bea5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
@@ -411,6 +411,10 @@ public class Analysis implements IAnalysis {
return groupingSets.containsKey(NodeRef.of(node));
}
+ public boolean containsAggregationQuery() {
+ return !groupingSets.isEmpty();
+ }
+
public GroupingSetAnalysis getGroupingSets(QuerySpecification node) {
return groupingSets.get(NodeRef.of(node));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 41714bdeb69..6c1becfcc9c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -524,10 +524,9 @@ public class TableDistributedPlanGenerator
}
List<PlanNode> childrenNodes = node.getChild().accept(this, context);
OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
- // TODO add back while implementing StreamingAggregationOperator
- // if (childOrdering != null) {
- // nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
- // }
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
if (childrenNodes.size() == 1) {
node.setChild(childrenNodes.get(0));
@@ -552,10 +551,9 @@ public class TableDistributedPlanGenerator
intermediate.getStep(),
intermediate.getHashSymbol(),
intermediate.getGroupIdSymbol());
- // TODO add back while implementing StreamingAggregationOperator
- // if (node.isStreamable()) {
- // nodeOrderingMap.put(planNodeId, childOrdering);
- // }
+ if (node.isStreamable()) {
+ nodeOrderingMap.put(planNodeId, childOrdering);
+ }
return aggregationNode;
})
.collect(Collectors.toList());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
index aed0bc0ba61..ecdef92b6b4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
@@ -56,7 +56,7 @@ public class PushAggregationIntoTableScan implements PlanOptimizer {
@Override
public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) {
if (!(context.getAnalysis().getStatement() instanceof Query)
- || context.getAnalysis().noAggregates()) {
+ || !context.getAnalysis().containsAggregationQuery()) {
return plan;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformAggregationToStreamable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformAggregationToStreamable.java
index 1ca5e2bbcd2..fb47681e9d0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformAggregationToStreamable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TransformAggregationToStreamable.java
@@ -50,7 +50,7 @@ public class TransformAggregationToStreamable implements PlanOptimizer {
@Override
public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) {
if (!(context.getAnalysis().getStatement() instanceof Query)
- || context.getAnalysis().noAggregates()) {
+ || !context.getAnalysis().containsAggregationQuery()) {
return plan;
}
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 9f8d4919ee8..0cc31381072 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
@@ -26,8 +26,6 @@ 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;
-import org.apache.tsfile.read.common.type.DateType;
import org.apache.tsfile.read.common.type.DoubleType;
import org.apache.tsfile.read.common.type.FloatType;
import org.apache.tsfile.read.common.type.RowType;
@@ -51,6 +49,8 @@ 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.BooleanType.BOOLEAN;
+import static org.apache.tsfile.read.common.type.DateType.DATE;
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;
@@ -97,7 +97,7 @@ public class TypeUtil {
public static Type getType(TypeEnum typeEnum, List<Type> subTypes) {
switch (typeEnum) {
case BOOLEAN:
- return BooleanType.BOOLEAN;
+ return BOOLEAN;
case INT32:
return INT32;
case INT64:
@@ -115,7 +115,7 @@ public class TypeUtil {
case TIMESTAMP:
return TimestampType.TIMESTAMP;
case DATE:
- return DateType.DATE;
+ return DATE;
case ROW:
return RowType.anonymous(subTypes);
default:
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AggregationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AggregationTest.java
index d658c83a41b..de711140a50 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AggregationTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AggregationTest.java
@@ -43,6 +43,7 @@ import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.exchange;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.expression;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.filter;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.mergeSort;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.output;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.project;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.singleGroupingSet;
@@ -152,7 +153,7 @@ public class AggregationTest {
ImmutableSet.of("time", "tag1", "tag2", "tag3", "s1", "s2"))))));
// - Exchange
- // Output - Project - Agg(FINAL) - Collect - Agg(PARTIAL) - TableScan
+ // Output - Project - Agg(FINAL) - MergeSort - Agg(PARTIAL) - TableScan
// - Exchange
assertPlan(
planTester.getFragmentPlan(0),
@@ -166,7 +167,7 @@ public class AggregationTest {
ImmutableList.of("tag1", "tag2", "tag3"), // Streamable
Optional.empty(),
FINAL,
- collect(
+ mergeSort(
exchange(),
aggregation(
singleGroupingSet("s1", "tag1", "tag2", "tag3", "time"),